并行解析模式下，获取meta逻辑存在线程安全问题

[lanxinxu]

• I have searched the issues of this repository and believe that this is not a duplicate.
• I have checked the FAQ of this repository and believe that this is not a duplicate.

• canal version
  1.1.2
• mysql version
  5.7

Issue Description

并行解析模式下，获取meta逻辑存在线程安全问题. connection是线程非安全的，当meta过期时会触发异常。
非tsdb模式逻辑如下，tsdb模式同理:

private TableMeta getTableMetaByDB(String fullname) throws IOException {
try {
ResultSetPacket packet = connection.query("show create table " + fullname);
String[] names = StringUtils.split(fullname, ".");
String schema = names[0];
String table = names[1].substring(0, names[1].length());
return new TableMeta(schema, table, parseTableMeta(schema, table, packet));
} catch (Throwable e) { // fallback to desc table
ResultSetPacket packet = connection.query("desc " + fullname);
String[] names = StringUtils.split(fullname, ".");
String schema = names[0];
String table = names[1].substring(0, names[1].length());
return new TableMeta(schema, table, parseTableMetaByDesc(packet));
}
}

异常栈如下：
2019-05-31 21:34:51.678 [destination =, address = , EventParser] ERROR com.alibaba.otter.canal.common.alarm.LogAlarmHandler - destination:*
[com.alibaba.otter.canal.parse.exception.CanalParseException: com.alibaba.otter.canal.parse.exception.CanalParseException: parse row data failed.
Caused by: com.alibaba.otter.canal.parse.exception.CanalParseException: parse row data failed.
Caused by: com.alibaba.otter.canal.parse.exception.CanalParseException: com.google.common.util.concurrent.UncheckedExecutionException: com.alibaba.otter.canal.parse.exception.CanalParseException: fetch failed by table meta:stress_test.d1
Caused by: com.google.common.util.concurrent.UncheckedExecutionException: com.alibaba.otter.canal.parse.exception.CanalParseException: fetch failed by table meta:stress_test.d1
at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2203)
at com.google.common.cache.LocalCache.get(LocalCache.java:3937)
at com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:3941)
at com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:4824)
at com.google.common.cache.LocalCache$LocalLoadingCache.getUnchecked(LocalCache.java:4830)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache.getTableMeta(TableMetaCache.java:196)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.getTableMeta(LogEventConvert.java:956)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parseRowsEventForTableMeta(LogEventConvert.java:504)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parseRowsEvent(LogEventConvert.java:525)
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor$DmlParserStage.onEvent(MysqlMultiStageCoprocessor.java:330)
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor$DmlParserStage.onEvent(MysqlMultiStageCoprocessor.java:316)
at com.lmax.disruptor.WorkProcessor.run(WorkProcessor.java:143)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)
Caused by: com.alibaba.otter.canal.parse.exception.CanalParseException: fetch failed by table meta:stress_test.d1
Caused by: java.io.IOException: should execute connector.connect() first
at com.alibaba.otter.canal.parse.driver.mysql.MysqlQueryExecutor.(MysqlQueryExecutor.java:30)
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlConnection.query(MysqlConnection.java:104)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache.getTableMetaByDB(TableMetaCache.java:93)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache.access$000(TableMetaCache.java:32)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache$1.load(TableMetaCache.java:63)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache$1.load(TableMetaCache.java:53)
at com.google.common.cache.LocalCache$LoadingValueReference.loadFuture(LocalCache.java:3527)
at com.google.common.cache.LocalCache$Segment.loadSync(LocalCache.java:2319)
at com.google.common.cache.LocalCache$Segment.lockedGetOrLoad(LocalCache.java:2282)
at com.google.common.cache.LocalCache$Segment.get(LocalCache.java:2197)
at com.google.common.cache.LocalCache.get(LocalCache.java:3937)
at com.google.common.cache.LocalCache.getOrLoad(LocalCache.java:3941)
at com.google.common.cache.LocalCache$LocalLoadingCache.get(LocalCache.java:4824)
at com.google.common.cache.LocalCache$LocalLoadingCache.getUnchecked(LocalCache.java:4830)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.TableMetaCache.getTableMeta(TableMetaCache.java:196)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.getTableMeta(LogEventConvert.java:956)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parseRowsEventForTableMeta(LogEventConvert.java:504)
at com.alibaba.otter.canal.parse.inbound.mysql.dbsync.LogEventConvert.parseRowsEvent(LogEventConvert.java:525)
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor$DmlParserStage.onEvent(MysqlMultiStageCoprocessor.java:330)
at com.alibaba.otter.canal.parse.inbound.mysql.MysqlMultiStageCoprocessor$DmlParserStage.onEvent(MysqlMultiStageCoprocessor.java:316)
at com.lmax.disruptor.WorkProcessor.run(WorkProcessor.java:143)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
at java.lang.Thread.run(Thread.java:745)

Steps to reproduce

制造meta过期场景即可。如：
binlog中的字段数量与mysql中的字段数量不一致，此时重启canal进程(非tsdb)

解决这个问题pr如下：
同步方式获取meta
#1866

@agapple

[luwave]

没有太理解线程安全问题

[lanxinxu]

那你找到这个方法getTableMetaByDB，看下具体实现吧。在并行解析模式下这个方法会被并发调用，而方法中用到的connection(封装了socket)并不是线程安全的，从而会有上述异常
@luwave

[agapple]

代码已合并