Merge pull request #20 from alibaba/master

merge

client-adapter/launcher/pom.xml

@@ -137,7 +137,7 @@
                         </goals>
                         <configuration>
                             <includeClassifiers>jar-with-dependencies</includeClassifiers>
-                            <outputDirectory>${project.basedir}/target/canal-adapter/lib</outputDirectory>
+                            <outputDirectory>${project.basedir}/target/canal-adapter/plugin</outputDirectory>
                         </configuration>
                     </execution>
                 </executions>

client-adapter/launcher/src/main/assembly/dev.xml

@@ -57,6 +57,10 @@
                 <exclude>**/*</exclude>
             </excludes>
         </fileSet>
+        <fileSet>
+            <directory>${project.basedir}/target/canal-adapter/plugin</directory>
+            <outputDirectory>/plugin/</outputDirectory>
+        </fileSet>
     </fileSets>
     <dependencySets>
         <dependencySet>

client-adapter/launcher/src/main/assembly/release.xml

@@ -58,13 +58,9 @@
                 <exclude>**/*</exclude>
             </excludes>
         </fileSet>
-
         <fileSet>
-            <directory>${project.basedir}/target/canal-adapter/lib</directory>
+            <directory>${project.basedir}/target/canal-adapter/plugin</directory>
-            <outputDirectory>/lib/</outputDirectory>
+            <outputDirectory>/plugin/</outputDirectory>
-            <includes>
-                <include>*-jar-with-dependencies.jar</include>
-            </includes>
         </fileSet>
     </fileSets>
     <dependencySets>

client-adapter/launcher/src/main/bin/startup.bat

@@ -8,7 +8,7 @@ if "%OS%" == "Windows_NT" set ENV_PATH=%~dp0%
 set conf_dir=%ENV_PATH%\..\conf
 
 set CLASSPATH=%conf_dir%
-set CLASSPATH=%conf_dir%\..\lib\*;%CLASSPATH%
+set CLASSPATH=%conf_dir%\..\lib\*;%conf_dir%\..\plugin\*;%CLASSPATH%
 
 set JAVA_MEM_OPTS= -Xms128m -Xmx512m -XX:PermSize=128m
 set JAVA_OPTS_EXT= -Djava.awt.headless=true -Djava.net.preferIPv4Stack=true -Dapplication.codeset=UTF-8 -Dfile.encoding=UTF-8

client-adapter/launcher/src/main/bin/startup.sh

@@ -53,6 +53,10 @@ ADAPTER_OPTS="-DappName=canal-adapter"
 for i in $base/lib/*;
     do CLASSPATH=$i:"$CLASSPATH";
 done
+
+for i in $base/plugin/*;
+    do CLASSPATH=$i:"$CLASSPATH";
+done
 CLASSPATH="$base/conf:$CLASSPATH";
 
 echo "cd to $bin_abs_path for workaround relative path"

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/MysqlEventParser.java

@@ -651,7 +651,7 @@ public class MysqlEventParser extends AbstractMysqlEventParser implements CanalE
                 throw new CanalParseException("command : 'show master status' has an error! pls check. you need (at least one of) the SUPER,REPLICATION CLIENT privilege(s) for this operation");
             }
             EntryPosition endPosition = new EntryPosition(fields.get(0), Long.valueOf(fields.get(1)));
-            if (isGTIDMode && fields.size() > 4) {
+            if (isGTIDMode() && fields.size() > 4) {
                 endPosition.setGtid(fields.get(4));
             }
             return endPosition;

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/LogEventConvert.java

@@ -646,15 +646,16 @@ public class LogEventConvert extends AbstractCanalLifeCycle implements BinlogPar
                 continue;
             }
 
-            if (fieldMeta != null && existOptionalMetaData) {
+            if (fieldMeta != null && existOptionalMetaData && tableMetaCache.isOnTSDB()) {
                 // check column info
                 boolean check = StringUtils.equalsIgnoreCase(fieldMeta.getColumnName(), info.name);
                 check &= (fieldMeta.isUnsigned() == info.unsigned);
                 check &= (fieldMeta.isNullable() == info.nullable);
 
                 if (!check) {
-                    throw new CanalParseException("MySQL8.0 unmatch column metadata & pls submit issue , db : "
+                    throw new CanalParseException("MySQL8.0 unmatch column metadata & pls submit issue , table : "
-                                                  + fieldMeta.toString() + " , binlog : " + info.toString()
+                                                  + tableMeta.getFullName() + ", db fieldMeta : "
+                                                  + fieldMeta.toString() + " , binlog fieldMeta : " + info.toString()
                                                   + " , on : " + event.getHeader().getLogFileName() + ":"
                                                   + event.getHeader().getLogPos());
                 }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/dbsync/TableMetaCache.java

@@ -39,6 +39,7 @@ public class TableMetaCache {
     public static final String              EXTRA          = "EXTRA";
     private MysqlConnection                 connection;
     private boolean                         isOnRDS        = false;
+    private boolean                         isOnTSDB       = false;
 
     private TableMetaTSDB                   tableMetaTSDB;
     // 第一层tableId,第二层schema.table,解决tableId重复，对应多张表
@@ -67,6 +68,8 @@ public class TableMetaCache {
                 }
 
             });
+        } else {
+            isOnTSDB = true;
         }
 
         try {
@@ -244,6 +247,15 @@ public class TableMetaCache {
             .toString();
     }
 
+
+    public boolean isOnTSDB() {
+        return isOnTSDB;
+    }
+
+    public void setOnTSDB(boolean isOnTSDB) {
+        this.isOnTSDB = isOnTSDB;
+    }
+
     public boolean isOnRDS() {
         return isOnRDS;
     }

parse/src/main/java/com/alibaba/otter/canal/parse/inbound/mysql/tsdb/MemoryTableMeta.java

@@ -213,6 +213,7 @@ public class MemoryTableMeta implements TableMetaTSDB {
                     fieldMeta.setNullable(true);
                 } else if (constraint instanceof SQLColumnPrimaryKey) {
                     fieldMeta.setKey(true);
+                    fieldMeta.setNullable(false);
                 } else if (constraint instanceof SQLColumnUniqueKey) {
                     fieldMeta.setUnique(true);
                 }
@@ -225,6 +226,7 @@ public class MemoryTableMeta implements TableMetaTSDB {
                 String name = getSqlName(pk.getExpr());
                 FieldMeta field = tableMeta.getFieldMetaByName(name);
                 field.setKey(true);
+                field.setNullable(false);
             }
         } else if (element instanceof MySqlUnique) {
             MySqlUnique column = (MySqlUnique) element;