From 34bcfbc3cb1c7ba992e4d248cca42babc7752a60 Mon Sep 17 00:00:00 2001 From: "jianhao.dai" Date: Thu, 20 Jul 2017 11:35:19 +0800 Subject: [PATCH 1/6] =?UTF-8?q?=E4=B8=8D=E7=AE=A1=E6=9C=89=E6=B2=A1?= =?UTF-8?q?=E6=9C=89filter=EF=BC=8C=E9=83=BD=E5=BA=94=E8=AF=A5=E5=AF=B9met?= =?UTF-8?q?a=20cache=E6=9B=B4=E6=96=B0=E4=B8=80=E4=B8=8B?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../inbound/mysql/dbsync/LogEventConvert.java | 40 +++++++++---------- 1 file changed, 20 insertions(+), 20 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index 1bbb0eb8..d1c3ca3b 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -164,6 +164,26 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar || result.getType() == EventType.RENAME || result.getType() == EventType.CINDEX || result.getType() == EventType.DINDEX) { // 针对DDL类型 + // 更新下table meta cache + if (tableMetaCache != null + && (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME)) { + for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) { + String schemaName0 = event.getDbName(); // 防止rename语句后产生schema变更带来影响 + if (StringUtils.isNotEmpty(renameResult.getSchemaName())) { + schemaName0 = renameResult.getSchemaName(); + } + + tableName = renameResult.getTableName(); + if (StringUtils.isNotEmpty(tableName)) { + // 如果解析到了正确的表信息，则根据全名进行清除 + tableMetaCache.clearTableMeta(schemaName0, tableName); + } else { + // 如果无法解析正确的表信息，则根据schema进行清除 + tableMetaCache.clearTableMetaWithSchemaName(schemaName0); + } + } + } + if (filterQueryDdl) { return null; } @@ -214,26 +234,6 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar return null; } - // 更新下table meta cache - if (tableMetaCache != null - && (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME)) { - for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) { - String schemaName0 = event.getDbName(); // 防止rename语句后产生schema变更带来影响 - if (StringUtils.isNotEmpty(renameResult.getSchemaName())) { - schemaName0 = renameResult.getSchemaName(); - } - - tableName = renameResult.getTableName(); - if (StringUtils.isNotEmpty(tableName)) { - // 如果解析到了正确的表信息，则根据全名进行清除 - tableMetaCache.clearTableMeta(schemaName0, tableName); - } else { - // 如果无法解析正确的表信息，则根据schema进行清除 - tableMetaCache.clearTableMetaWithSchemaName(schemaName0); - } - } - } - Header header = createHeader(binlogFileName, event.getHeader(), schemaName, tableName, type); RowChange.Builder rowChangeBuider = RowChange.newBuilder(); if (result.getType() != EventType.QUERY) { From 214e404e1b10f20d16cb50fa9c1c8488dfc719bd Mon Sep 17 00:00:00 2001 From: "jianhao.dai" Date: Thu, 20 Jul 2017 11:37:24 +0800 Subject: [PATCH 2/6] =?UTF-8?q?=E6=8C=AA=E5=88=B0=E6=9C=80=E5=89=8D?= =?UTF-8?q?=E9=9D=A2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../inbound/mysql/dbsync/LogEventConvert.java | 41 ++++++++++--------- 1 file changed, 21 insertions(+), 20 deletions(-) diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java index d1c3ca3b..cdc7d7fe 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java @@ -158,32 +158,33 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar String tableName = result.getTableName(); EventType type = EventType.QUERY; + + // 更新下table meta cache + if (tableMetaCache != null + && (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME)) { + for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) { + String schemaName0 = event.getDbName(); // 防止rename语句后产生schema变更带来影响 + if (StringUtils.isNotEmpty(renameResult.getSchemaName())) { + schemaName0 = renameResult.getSchemaName(); + } + + tableName = renameResult.getTableName(); + if (StringUtils.isNotEmpty(tableName)) { + // 如果解析到了正确的表信息，则根据全名进行清除 + tableMetaCache.clearTableMeta(schemaName0, tableName); + } else { + // 如果无法解析正确的表信息，则根据schema进行清除 + tableMetaCache.clearTableMetaWithSchemaName(schemaName0); + } + } + } + // fixed issue https://github.com/alibaba/canal/issues/58 if (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.CREATE || result.getType() == EventType.TRUNCATE || result.getType() == EventType.RENAME || result.getType() == EventType.CINDEX || result.getType() == EventType.DINDEX) { // 针对DDL类型 - // 更新下table meta cache - if (tableMetaCache != null - && (result.getType() == EventType.ALTER || result.getType() == EventType.ERASE || result.getType() == EventType.RENAME)) { - for (DdlResult renameResult = result; renameResult != null; renameResult = renameResult.getRenameTableResult()) { - String schemaName0 = event.getDbName(); // 防止rename语句后产生schema变更带来影响 - if (StringUtils.isNotEmpty(renameResult.getSchemaName())) { - schemaName0 = renameResult.getSchemaName(); - } - - tableName = renameResult.getTableName(); - if (StringUtils.isNotEmpty(tableName)) { - // 如果解析到了正确的表信息，则根据全名进行清除 - tableMetaCache.clearTableMeta(schemaName0, tableName); - } else { - // 如果无法解析正确的表信息，则根据schema进行清除 - tableMetaCache.clearTableMetaWithSchemaName(schemaName0); - } - } - } - if (filterQueryDdl) { return null; } From 67436b2bc4a8e23faabb1555c0d34a2f53e060dc Mon Sep 17 00:00:00 2001 From: agapple Date: Thu, 3 Aug 2017 21:15:09 +0800 Subject: [PATCH 3/6] fixed issue #330 & #350 , json length & escapse --- .../tddl/dbsync/binlog/JsonConversion.java | 19 +++++++++++++-- .../dbsync/binlog/event/RowsLogBuffer.java | 24 +++++++++++++++++-- .../parse/inbound/AbstractEventParser.java | 2 +- 3 files changed, 40 insertions(+), 5 deletions(-) diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java index 251e458f..020591f7 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/JsonConversion.java @@ -363,12 +363,12 @@ public class JsonConversion { buf.append('"').append(text).append('"'); } else { text = m_data.getFixString((int) m_length, charsetName); - buf.append('"').append(text).append('"'); + buf.append('"').append(escapse(text)).append('"'); } break; case STRING: - buf.append('"').append(m_string_value).append('"'); + buf.append('"').append(escapse(m_string_value)).append('"'); break; case ERROR: throw new IllegalArgumentException("illegal json data"); @@ -378,6 +378,21 @@ public class JsonConversion { } } + private static StringBuilder escapse(String data) { + StringBuilder sb = new StringBuilder(data.length()); + int endIndex = data.length(); + for (int i = 0; i < endIndex; ++i) { + char c = data.charAt(i); + if (c == '"') { + sb.append('\\'); + } else if (c == '\\') { + sb.append("\\"); + } + sb.append(c); + } + return sb; + } + public static enum Json_enum_type { OBJECT, ARRAY, STRING, INT, UINT, DOUBLE, LITERAL_NULL, LITERAL_TRUE, LITERAL_FALSE, OPAQUE, ERROR } diff --git a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java index 2456f358..459da9cc 100644 --- a/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java +++ b/dbsync/src/main/java/com/taobao/tddl/dbsync/binlog/event/RowsLogBuffer.java @@ -949,8 +949,28 @@ public final class RowsLogBuffer { break; } case LogEvent.MYSQL_TYPE_JSON: { - len = buffer.getUint16(); - buffer.forward(meta - 2); + switch (meta) { + case 1: { + len = buffer.getUint8(); + break; + } + case 2: { + len = buffer.getUint16(); + break; + } + case 3: { + len = buffer.getUint24(); + break; + } + case 4: { + len = (int) buffer.getUint32(); + break; + } + default: + throw new IllegalArgumentException("!! Unknown JSON packlen = " + meta); + } + // len = buffer.getUint16(); + // buffer.forward(meta - 4); int position = buffer.position(); Json_Value jsonValue = JsonConversion.parse_value(buffer.getUint8(), buffer, len - 1, charsetName); StringBuilder builder = new StringBuilder(); diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index 6b34edf5..18f8c9f8 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -275,7 +275,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle parseThread.setUncaughtExceptionHandler(handler); parseThread.setName(String.format("destination = %s , address = %s , EventParser", destination, - runningInfo == null ? null : runningInfo.getAddress().toString())); + runningInfo == null ? null : runningInfo.getAddress())); parseThread.start(); } From a1cf292baa13c8ae2586160d1525277055635afb Mon Sep 17 00:00:00 2001 From: "alex.zheng" Date: Tue, 29 Aug 2017 16:03:54 +0800 Subject: [PATCH 4/6] try to avoid use multiline exception directly --- .../client/impl/ClusterCanalConnector.java | 48 ++++++++----------- .../otter/canal/deployer/CanalLauncher.java | 7 +-- .../monitor/SpringInstanceConfigMonitor.java | 10 ++-- .../parse/driver/mysql/MysqlConnector.java | 7 ++- .../canal/example/ClusterCanalClientTest.java | 4 +- .../canal/example/SimpleCanalClientTest.java | 4 +- .../parse/inbound/AbstractEventParser.java | 2 +- .../parse/inbound/mysql/MysqlConnection.java | 15 +++--- .../parse/inbound/mysql/MysqlEventParser.java | 12 ++--- 9 files changed, 44 insertions(+), 65 deletions(-) diff --git a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java index 9fd1ca10..1c4027c7 100644 --- a/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java +++ b/client/src/main/java/com/alibaba/otter/canal/client/impl/ClusterCanalConnector.java @@ -3,7 +3,6 @@ package com.alibaba.otter.canal.client.impl; import java.net.SocketAddress; import java.util.concurrent.TimeUnit; -import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -107,9 +106,10 @@ public class ClusterCanalConnector implements CanalConnector { this.filter = filter; return; } catch (Throwable t) { - logger.warn("something goes wrong when subscribing from server:{}\n{}", - currentConnector != null ? currentConnector.getAddress() : "null", - ExceptionUtils.getFullStackTrace(t)); + logger.warn(String.format( + "something goes wrong when subscribing from server: %s", + currentConnector != null ? currentConnector.getAddress() : "null"), + t); times++; restart(); logger.info("restart the connector for next round retry."); @@ -126,9 +126,8 @@ public class ClusterCanalConnector implements CanalConnector { currentConnector.unsubscribe(); return; } catch (Throwable t) { - logger.warn("something goes wrong when unsubscribing from server:{}\n{}", - currentConnector != null ? currentConnector.getAddress() : "null", - ExceptionUtils.getFullStackTrace(t)); + logger.warn(String.format("something goes wrong when unsubscribing from server:%s", + currentConnector != null ? currentConnector.getAddress() : "null"), t); times++; restart(); logger.info("restart the connector for next round retry."); @@ -144,9 +143,8 @@ public class ClusterCanalConnector implements CanalConnector { Message msg = currentConnector.get(batchSize); return msg; } catch (Throwable t) { - logger.warn("something goes wrong when getting data from server:{}\n{}", - currentConnector != null ? currentConnector.getAddress() : "null", - ExceptionUtils.getFullStackTrace(t)); + logger.warn(String.format("something goes wrong when getting data from server:%s", + currentConnector != null ? currentConnector.getAddress() : "null"), t); times++; restart(); logger.info("restart the connector for next round retry."); @@ -162,9 +160,8 @@ public class ClusterCanalConnector implements CanalConnector { Message msg = currentConnector.get(batchSize, timeout, unit); return msg; } catch (Throwable t) { - logger.warn("something goes wrong when getting data from server:{}\n{}", - currentConnector != null ? currentConnector.getAddress() : "null", - ExceptionUtils.getFullStackTrace(t)); + logger.warn(String.format("something goes wrong when getting data from server:%s", + currentConnector != null ? currentConnector.getAddress() : "null"), t); times++; restart(); logger.info("restart the connector for next round retry."); @@ -180,9 +177,8 @@ public class ClusterCanalConnector implements CanalConnector { Message msg = currentConnector.getWithoutAck(batchSize); return msg; } catch (Throwable t) { - logger.warn("something goes wrong when getWithoutAck data from server:{}\n{}", - currentConnector.getAddress(), - ExceptionUtils.getFullStackTrace(t)); + logger.warn(String.format("something goes wrong when getWithoutAck data from server:%s", + currentConnector.getAddress()), t); times++; restart(); logger.info("restart the connector for next round retry."); @@ -198,9 +194,8 @@ public class ClusterCanalConnector implements CanalConnector { Message msg = currentConnector.getWithoutAck(batchSize, timeout, unit); return msg; } catch (Throwable t) { - logger.warn("something goes wrong when getWithoutAck data from server:{}\n{}", - currentConnector.getAddress(), - ExceptionUtils.getFullStackTrace(t)); + logger.warn(String.format("something goes wrong when getWithoutAck data from server:%s", + currentConnector.getAddress()), t); times++; restart(); logger.info("restart the connector for next round retry."); @@ -216,9 +211,8 @@ public class ClusterCanalConnector implements CanalConnector { currentConnector.rollback(batchId); return; } catch (Throwable t) { - logger.warn("something goes wrong when rollbacking data from server:{}\n{}", - currentConnector.getAddress(), - ExceptionUtils.getFullStackTrace(t)); + logger.warn(String.format("something goes wrong when rollbacking data from server:%s", + currentConnector.getAddress()), t); times++; restart(); logger.info("restart the connector for next round retry."); @@ -234,9 +228,8 @@ public class ClusterCanalConnector implements CanalConnector { currentConnector.rollback(); return; } catch (Throwable t) { - logger.warn("something goes wrong when rollbacking data from server:{}\n{}", - currentConnector.getAddress(), - ExceptionUtils.getFullStackTrace(t)); + logger.warn(String.format("something goes wrong when rollbacking data from server:%s", + currentConnector.getAddress()), t); times++; restart(); logger.info("restart the connector for next round retry."); @@ -253,9 +246,8 @@ public class ClusterCanalConnector implements CanalConnector { currentConnector.ack(batchId); return; } catch (Throwable t) { - logger.warn("something goes wrong when acking data from server:{}\n{}", - currentConnector.getAddress(), - ExceptionUtils.getFullStackTrace(t)); + logger.warn(String.format("something goes wrong when acking data from server:%s", + currentConnector.getAddress()), t); times++; restart(); logger.info("restart the connector for next round retry."); diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index e0d5d7ab..63119eef 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -4,7 +4,6 @@ import java.io.FileInputStream; import java.util.Properties; import org.apache.commons.lang.StringUtils; -import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -41,8 +40,7 @@ public class CanalLauncher { logger.info("## stop the canal server"); controller.stop(); } catch (Throwable e) { - logger.warn("##something goes wrong when stopping canal Server:\n{}", - ExceptionUtils.getFullStackTrace(e)); + logger.warn("##something goes wrong when stopping canal Server:", e); } finally { logger.info("## canal server is down."); } @@ -50,8 +48,7 @@ public class CanalLauncher { }); } catch (Throwable e) { - logger.error("## Something goes wrong when starting up the canal Server:\n{}", - ExceptionUtils.getFullStackTrace(e)); + logger.error("## Something goes wrong when starting up the canal Server:", e); System.exit(0); } } diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java index 864753ab..38d6d24c 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/monitor/SpringInstanceConfigMonitor.java @@ -13,7 +13,6 @@ import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; import org.apache.commons.lang.StringUtils; -import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.util.Assert; @@ -186,7 +185,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen logger.info("auto notify start {} successful.", destination); } catch (Throwable e) { - logger.error("scan add found[{}] but start failed", destination, ExceptionUtils.getFullStackTrace(e)); + logger.error(String.format("scan add found[%s] but start failed", destination), e); } } @@ -197,7 +196,7 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen lastFiles.remove(destination); logger.info("auto notify stop {} successful.", destination); } catch (Throwable e) { - logger.error("scan delete found[{}] but stop failed", destination, ExceptionUtils.getFullStackTrace(e)); + logger.error(String.format("scan delete found[%s] but stop failed", destination), e); actions.put(destination, action);// 再重新加回去，下一次scan时再执行删除 } } @@ -209,9 +208,8 @@ public class SpringInstanceConfigMonitor extends AbstractCanalLifeCycle implemen action.reload(destination); logger.info("auto notify reload {} successful.", destination); } catch (Throwable e) { - logger.error("scan reload found[{}] but reload failed", - destination, - ExceptionUtils.getFullStackTrace(e)); + logger.error(String.format("scan reload found[%s] but reload failed", + destination), e); } } } diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java index a05c3b19..a5f89d83 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/MysqlConnector.java @@ -4,7 +4,6 @@ import java.io.IOException; import java.net.InetSocketAddress; import java.util.concurrent.atomic.AtomicBoolean; -import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -69,7 +68,7 @@ public class MysqlConnector { negotiate(channel); } catch (Exception e) { disconnect(); - throw new IOException("connect " + this.address + " failure:" + ExceptionUtils.getStackTrace(e)); + throw new IOException("connect " + this.address + " failure", e); } } else { logger.error("the channel can't be connected twice."); @@ -89,7 +88,7 @@ public class MysqlConnector { } logger.info("disConnect MysqlConnection to {}...", address); } catch (Exception e) { - throw new IOException("disconnect " + this.address + " failure:" + ExceptionUtils.getStackTrace(e)); + throw new IOException("disconnect " + this.address + " failure", e); } // 执行一次quit @@ -101,7 +100,7 @@ public class MysqlConnector { MysqlUpdateExecutor executor = new MysqlUpdateExecutor(connector); executor.update("KILL CONNECTION " + connectionId); } catch (Exception e) { - throw new IOException("KILL DUMP " + connectionId + " failure:" + ExceptionUtils.getStackTrace(e)); + throw new IOException("KILL DUMP " + connectionId + " failure", e); } finally { if (connector != null) { connector.disconnect(); diff --git a/example/src/main/java/com/alibaba/otter/canal/example/ClusterCanalClientTest.java b/example/src/main/java/com/alibaba/otter/canal/example/ClusterCanalClientTest.java index 478cc304..d4a18af0 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/ClusterCanalClientTest.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/ClusterCanalClientTest.java @@ -1,7 +1,5 @@ package com.alibaba.otter.canal.example; -import org.apache.commons.lang.exception.ExceptionUtils; - import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; @@ -41,7 +39,7 @@ public class ClusterCanalClientTest extends AbstractCanalClientTest { logger.info("## stop the canal client"); clientTest.stop(); } catch (Throwable e) { - logger.warn("##something goes wrong when stopping canal:\n{}", ExceptionUtils.getFullStackTrace(e)); + logger.warn("##something goes wrong when stopping canal:", e); } finally { logger.info("## canal client is down."); } diff --git a/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java b/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java index 9fa8576d..658a1419 100644 --- a/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java +++ b/example/src/main/java/com/alibaba/otter/canal/example/SimpleCanalClientTest.java @@ -2,8 +2,6 @@ package com.alibaba.otter.canal.example; import java.net.InetSocketAddress; -import org.apache.commons.lang.exception.ExceptionUtils; - import com.alibaba.otter.canal.client.CanalConnector; import com.alibaba.otter.canal.client.CanalConnectors; import com.alibaba.otter.canal.common.utils.AddressUtils; @@ -39,7 +37,7 @@ public class SimpleCanalClientTest extends AbstractCanalClientTest { logger.info("## stop the canal client"); clientTest.stop(); } catch (Throwable e) { - logger.warn("##something goes wrong when stopping canal:\n{}", ExceptionUtils.getFullStackTrace(e)); + logger.warn("##something goes wrong when stopping canal:", e); } finally { logger.info("## canal client is down."); } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java index 18f8c9f8..7a423ae6 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/AbstractEventParser.java @@ -430,7 +430,7 @@ public abstract class AbstractEventParser extends AbstractCanalLifeCycle } } catch (Throwable e) { - logger.warn("heartBeat run failed " + ExceptionUtils.getStackTrace(e)); + logger.warn("heartBeat run failed ", e); } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java index c648c10d..7a7b4ad5 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlConnection.java @@ -6,7 +6,6 @@ import java.nio.charset.Charset; import java.util.List; import org.apache.commons.lang.StringUtils; -import org.apache.commons.lang.exception.ExceptionUtils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -167,25 +166,25 @@ public class MysqlConnection implements ErosaConnection { try { update("set wait_timeout=9999999"); } catch (Exception e) { - logger.warn(ExceptionUtils.getFullStackTrace(e)); + logger.warn("update wait_timeout failed", e); } try { update("set net_write_timeout=1800"); } catch (Exception e) { - logger.warn(ExceptionUtils.getFullStackTrace(e)); + logger.warn("update net_write_timeout failed", e); } try { update("set net_read_timeout=1800"); } catch (Exception e) { - logger.warn(ExceptionUtils.getFullStackTrace(e)); + logger.warn("update net_read_timeout failed", e); } try { // 设置服务端返回结果时不做编码转化，直接按照数据库的二进制编码进行发送，由客户端自己根据需求进行编码转化 update("set names 'binary'"); } catch (Exception e) { - logger.warn(ExceptionUtils.getFullStackTrace(e)); + logger.warn("update names failed", e); } try { @@ -195,7 +194,7 @@ public class MysqlConnection implements ErosaConnection { // 但也不能乱设置，需要和mysql server的checksum配置一致，不然RotateLogEvent会出现乱码 update("set @master_binlog_checksum= '@@global.binlog_checksum'"); } catch (Exception e) { - logger.warn(ExceptionUtils.getFullStackTrace(e)); + logger.warn("update master_binlog_checksum failed", e); } try { @@ -204,7 +203,7 @@ public class MysqlConnection implements ErosaConnection { update("set @slave_uuid=uuid()"); } catch (Exception e) { if (!StringUtils.contains(e.getMessage(), "Unknown system variable")) { - logger.warn(ExceptionUtils.getFullStackTrace(e)); + logger.warn("update slave_uuid failed", e); } } @@ -212,7 +211,7 @@ public class MysqlConnection implements ErosaConnection { // mariadb针对特殊的类型，需要设置session变量 update("SET @mariadb_slave_capability='" + LogEvent.MARIA_SLAVE_CAPABILITY_MINE + "'"); } catch (Exception e) { - logger.warn(ExceptionUtils.getFullStackTrace(e)); + logger.warn("update mariadb_slave_capability failed", e); } } diff --git a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java index 9b7b152a..fde9cf05 100644 --- a/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java +++ b/parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java @@ -10,7 +10,6 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import org.apache.commons.lang.StringUtils; -import org.apache.commons.lang.exception.ExceptionUtils; import org.springframework.util.CollectionUtils; import com.alibaba.otter.canal.common.utils.JsonUtils; @@ -237,19 +236,19 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE ((HeartBeatCallback) haController).onFailed(e); } reconnect = true; - logger.warn("connect failed by " + ExceptionUtils.getStackTrace(e)); + logger.warn("connect failed by ", e); } catch (IOException e) { if (haController != null && haController instanceof HeartBeatCallback) { ((HeartBeatCallback) haController).onFailed(e); } reconnect = true; - logger.warn("connect failed by " + ExceptionUtils.getStackTrace(e)); + logger.warn("connect failed by ", e); } catch (Throwable e) { if (haController != null && haController instanceof HeartBeatCallback) { ((HeartBeatCallback) haController).onFailed(e); } reconnect = true; - logger.warn("connect failed by " + ExceptionUtils.getStackTrace(e)); + logger.warn("connect failed by ", e); } } @@ -549,9 +548,8 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE return entryPosition; } } catch (Exception e) { - logger.warn("the binlogfile:{} doesn't exist, to continue to search the next binlogfile , caused by {}", - startSearchBinlogFile, - ExceptionUtils.getFullStackTrace(e)); + logger.warn(String.format("the binlogfile:%s doesn't exist, to continue to search the next binlogfile , caused by", + startSearchBinlogFile), e); int binlogSeqNum = Integer.parseInt(startSearchBinlogFile.substring(startSearchBinlogFile.indexOf(".") + 1)); if (binlogSeqNum <= 1) { logger.warn("Didn't find the corresponding binlog files"); From 041cbf04c7f4892dba4c87aa0d6141384ead45d3 Mon Sep 17 00:00:00 2001 From: "alex.zheng" Date: Wed, 6 Sep 2017 12:07:50 +0800 Subject: [PATCH 5/6] add uncaughtexception handler --- .../alibaba/otter/canal/deployer/CanalLauncher.java | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java index 63119eef..e66a5a9b 100644 --- a/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java +++ b/deployer/src/main/java/com/alibaba/otter/canal/deployer/CanalLauncher.java @@ -20,6 +20,10 @@ public class CanalLauncher { public static void main(String[] args) throws Throwable { try { + logger.info("## set default uncaught exception handler"); + setGlobalUncaughtExceptionHandler(); + + logger.info("## load canal configurations"); String conf = System.getProperty("canal.conf", "classpath:canal.properties"); Properties properties = new Properties(); if (conf.startsWith(CLASSPATH_URL_PREFIX)) { @@ -52,4 +56,13 @@ public class CanalLauncher { System.exit(0); } } + + private static void setGlobalUncaughtExceptionHandler() { + Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { + @Override + public void uncaughtException(Thread t, Throwable e) { + logger.error("UnCaughtException", e); + } + }); + } } From 788c1d9b84e470d28eaf1198ad0a5d7f18da287b Mon Sep 17 00:00:00 2001 From: agapple Date: Sat, 23 Sep 2017 11:50:25 +0800 Subject: [PATCH 6/6] fixed issue #360 , socket pool bug --- .../mysql/socket/SocketChannelPool.java | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java index 13efbed9..52609b19 100644 --- a/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java +++ b/driver/src/main/java/com/alibaba/otter/canal/parse/driver/mysql/socket/SocketChannelPool.java @@ -21,14 +21,16 @@ import java.net.SocketAddress; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; +import com.alibaba.otter.canal.common.utils.BooleanMutex; + /** * @author luoyaogui 实现channel的管理（监听连接、读数据、回收） 2016-12-28 */ @SuppressWarnings({ "rawtypes", "deprecation" }) public abstract class SocketChannelPool { - private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组 - private static Bootstrap boot = new Bootstrap(); // 主 + private static EventLoopGroup group = new NioEventLoopGroup(); // 非阻塞IO线程组 + private static Bootstrap boot = new Bootstrap(); // 主 private static Map chManager = new ConcurrentHashMap(); static { @@ -53,19 +55,20 @@ public abstract class SocketChannelPool { public static SocketChannel open(SocketAddress address) throws Exception { final SocketChannel socket = new SocketChannel(); + final BooleanMutex mutex = new BooleanMutex(false); boot.connect(address).addListener(new ChannelFutureListener() { @Override public void operationComplete(ChannelFuture arg0) throws Exception { - if (arg0.isSuccess()) socket.setChannel(arg0.channel()); - synchronized (socket) { - socket.notify(); + if (arg0.isSuccess()) { + socket.setChannel(arg0.channel()); } + + mutex.set(true); } }); - synchronized (socket) { - socket.wait(); - } + // wait for complete + mutex.get(); if (null == socket.getChannel()) { throw new IOException("can't create socket!"); }