Merge pull request #1112 from rewerma/master

client 排除jsr305不打进fat jar

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfig.java

@@ -76,6 +76,7 @@ public class MappingConfig {
     public static class ColumnItem {
 
         private boolean isRowKey = false;
+        private Integer rowKeyLen;
         private String  column;
         private String  family;
         private String  qualifier;
@@ -89,6 +90,14 @@ public class MappingConfig {
             isRowKey = rowKey;
         }
 
+        public Integer getRowKeyLen() {
+            return rowKeyLen;
+        }
+
+        public void setRowKeyLen(Integer rowKeyLen) {
+            this.rowKeyLen = rowKeyLen;
+        }
+
         public String getColumn() {
             return column;
         }
@@ -264,7 +273,16 @@ public class MappingConfig {
                     ColumnItem columnItem = new ColumnItem();
                     columnItem.setColumn(columnField.getKey());
                     columnItem.setType(type);
-                    if ("rowKey".equalsIgnoreCase(field)) {
+                    if (field != null && field.toUpperCase().startsWith("ROWKEY")) {
+                        int idx = field.toUpperCase().indexOf("LEN:");
+                        if (idx > -1) {
+                            String len = field.substring(idx + 4);
+                            try {
+                                columnItem.setRowKeyLen(Integer.parseInt(len));
+                            } catch (Exception e) {
+                                // ignore
+                            }
+                        }
                         columnItem.setRowKey(true);
                         rowKeyColumn = columnItem;
                     } else {

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseEtlService.java

@@ -199,8 +199,8 @@ public class HbaseEtlService {
                     } else {
                         sqlFinal = sql + " LIMIT " + offset + "," + cnt;
                     }
-                    Future<Boolean> future = executor
+                    Future<Boolean> future = executor.submit(
-                        .submit(() -> executeSqlImport(ds, sqlFinal, hbaseMapping, hbaseTemplate, successCount, errMsg));
+                        () -> executeSqlImport(ds, sqlFinal, hbaseMapping, hbaseTemplate, successCount, errMsg));
                     futures.add(future);
                 }
 
@@ -213,8 +213,8 @@ public class HbaseEtlService {
                 executeSqlImport(ds, sql, hbaseMapping, hbaseTemplate, successCount, errMsg);
             }
 
-            logger.info(
+            logger.info(hbaseMapping.getHbaseTable() + " etl completed in: "
-                hbaseMapping.getHbaseTable() + " etl completed in: " + (System.currentTimeMillis() - start) / 1000 + "s!");
+                        + (System.currentTimeMillis() - start) / 1000 + "s!");
 
             etlResult.setResultMessage("导入HBase表 " + hbaseMapping.getHbaseTable() + " 数据：" + successCount.get() + " 条");
         } catch (Exception e) {
@@ -332,11 +332,41 @@ public class HbaseEtlService {
 
                                     byte[] valBytes = Bytes.toBytes(val.toString());
                                     if (columnItem.isRowKey()) {
+                                        if (columnItem.getRowKeyLen() != null) {
+                                            valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val));
                                             row.setRowKey(valBytes);
+                                        } else {
+                                            row.setRowKey(valBytes);
+                                        }
                                     } else {
                                         row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes);
                                     }
                                 } else {
+                                    if (MappingConfig.Mode.STRING == hbaseMapping.getMode()) {
+                                        byte[] valBytes = Bytes.toBytes(val.toString());
+                                        if (columnItem.isRowKey()) {
+                                            if (columnItem.getRowKeyLen() != null) {
+                                                valBytes = Bytes.toBytes(limitLenNum(columnItem.getRowKeyLen(), val));
+                                            }
+                                            row.setRowKey(valBytes);
+                                        } else {
+                                            row.addCell(columnItem.getFamily(), columnItem.getQualifier(), valBytes);
+                                        }
+                                    } else if (MappingConfig.Mode.NATIVE == hbaseMapping.getMode()) {
+                                        Type type = Type.getType(columnItem.getType());
+                                        if (columnItem.isRowKey()) {
+                                            if (columnItem.getRowKeyLen() != null) {
+                                                String v = limitLenNum(columnItem.getRowKeyLen(), val);
+                                                row.setRowKey(Bytes.toBytes(v));
+                                            } else {
+                                                row.setRowKey(TypeUtil.toBytes(val, type));
+                                            }
+                                        } else {
+                                            row.addCell(columnItem.getFamily(),
+                                                columnItem.getQualifier(),
+                                                TypeUtil.toBytes(val, type));
+                                        }
+                                    } else if (MappingConfig.Mode.PHOENIX == hbaseMapping.getMode()) {
                                         PhType phType = PhType.getType(columnItem.getType());
                                         if (columnItem.isRowKey()) {
                                             row.setRowKey(PhTypeUtil.toBytes(val, phType));
@@ -348,6 +378,7 @@ public class HbaseEtlService {
                                     }
                                 }
                             }
+                        }
 
                         if (row.getRowKey() == null) throw new RuntimeException("RowKey 值为空");
 
@@ -382,4 +413,16 @@ public class HbaseEtlService {
             return false;
         }
     }
+
+    private static String limitLenNum(int len, Object val) {
+        if (val == null) {
+            return null;
+        }
+        if (val instanceof Number) {
+            return String.format("%0" + len + "d", (Number) ((Number) val).longValue());
+        } else if (val instanceof String) {
+            return String.format("%0" + len + "d", Long.parseLong((String) val));
+        }
+        return null;
+    }
 }

client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java

@@ -132,7 +132,21 @@ public class HbaseSyncService {
                     }
                 } else {
                     if (columnItem.isRowKey()) {
-                        // row.put("rowKey", bytes);
+                        if (columnItem.getRowKeyLen() != null && entry.getValue() != null) {
+                            if (entry.getValue() instanceof Number) {
+                                String v = String.format("%0" + columnItem.getRowKeyLen() + "d",
+                                    ((Number) entry.getValue()).longValue());
+                                bytes = Bytes.toBytes(v);
+                            } else {
+                                try {
+                                    String v = String.format("%0" + columnItem.getRowKeyLen() + "d",
+                                        Integer.parseInt((String) entry.getValue()));
+                                    bytes = Bytes.toBytes(v);
+                                } catch (Exception e) {
+                                    // ignore
+                                }
+                            }
+                        }
                         hRow.setRowKey(bytes);
                     } else {
                         hRow.addCell(columnItem.getFamily(), columnItem.getQualifier(), bytes);
@@ -146,8 +160,8 @@ public class HbaseSyncService {
     /**
      * 更新操作
      * 
-     * @param config
+     * @param config 配置对象
-     * @param dml
+     * @param dml dml对象
      */
     private void update(MappingConfig config, Dml dml) {
         List<Map<String, Object>> data = dml.getData();
@@ -192,7 +206,7 @@ public class HbaseSyncService {
                 Map<String, Object> rowKey = data.get(0);
                 rowKeyBytes = typeConvert(null, hbaseMapping, rowKey.values().iterator().next());
             } else {
-                rowKeyBytes = typeConvert(rowKeyColumn, hbaseMapping, r.get(rowKeyColumn.getColumn()));
+                rowKeyBytes = getRowKeyBytes(hbaseMapping, rowKeyColumn, r);
             }
             if (rowKeyBytes == null) throw new RuntimeException("rowKey值为空");
 
@@ -277,8 +291,7 @@ public class HbaseSyncService {
                 Map<String, Object> rowKey = data.get(0);
                 rowKeyBytes = typeConvert(null, hbaseMapping, rowKey.values().iterator().next());
             } else {
-                Object val = r.get(rowKeyColumn.getColumn());
+                rowKeyBytes = getRowKeyBytes(hbaseMapping, rowKeyColumn, r);
-                rowKeyBytes = typeConvert(rowKeyColumn, hbaseMapping, val);
             }
             if (rowKeyBytes == null) throw new RuntimeException("rowKey值为空");
             rowKeys.add(rowKeyBytes);
@@ -424,4 +437,22 @@ public class HbaseSyncService {
         return rowKeyValue.toString();
     }
 
+    private static byte[] getRowKeyBytes(MappingConfig.HbaseMapping hbaseMapping, MappingConfig.ColumnItem rowKeyColumn,
+                                         Map<String, Object> rowData) {
+        Object val = rowData.get(rowKeyColumn.getColumn());
+        String v = null;
+        if (rowKeyColumn.getRowKeyLen() != null) {
+            if (val instanceof Number) {
+                v = String.format("%0" + rowKeyColumn.getRowKeyLen() + "d", (Number) ((Number) val).longValue());
+            } else if (val instanceof String) {
+                v = String.format("%0" + rowKeyColumn.getRowKeyLen() + "d", Long.parseLong((String) val));
+            }
+        }
+        if (v != null) {
+            return Bytes.toBytes(v);
+        } else {
+            return typeConvert(rowKeyColumn, hbaseMapping, val);
+        }
+    }
+
 }

client-adapter/hbase/src/main/resources/hbase/mytest_person2.yml

@@ -1,7 +1,7 @@
 dataSourceKey: defaultDS
 destination: example
 hbaseMapping:
-  mode: PHOENIX  #NATIVE   #STRING
+  mode: STRING  #NATIVE   #PHOENIX
   database: mytest  # 数据库名
   table: person2     # 数据库表名
   hbaseTable: MYTEST.PERSON2   # HBase表名
@@ -11,14 +11,14 @@ hbaseMapping:
   #rowKey: id,type  # 复合字段rowKey不能和columns中的rowKey重复
   columns:
     # 数据库字段:HBase对应字段
-    id: ROWKEY$UNSIGNED_LONG
+    id: ROWKEY LEN:15
     name: NAME
     email: EMAIL
-    type: $DECIMAL
+    type:
-    c_time: C_TIME$UNSIGNED_TIMESTAMP
+    c_time: C_TIME
-    birthday: BIRTHDAY$DATE
+    birthday: BIRTHDAY
-  excludeColumns:
+#  excludeColumns:
-    - lat   # 忽略字段
+#    - lat   # 忽略字段
 
 # -- NATIVE类型
 # $DEFAULT

client-adapter/launcher/pom.xml

@@ -216,7 +216,7 @@
                             <descriptors>
                                 <descriptor>${basedir}/src/main/assembly/release.xml</descriptor>
                             </descriptors>
-                            <finalName>${project.artifactId}-${project.version}</finalName>
+                            <finalName>canal.adapter-${project.version}</finalName>
                             <outputDirectory>${project.basedir}/../../target</outputDirectory>
                         </configuration>
                     </plugin>

client-adapter/launcher/src/main/java/com/alibaba/otter/canal/adapter/launcher/rest/CommonRest.java

@@ -56,8 +56,11 @@ public class CommonRest {
     @PostMapping("/etl/{type}/{task}")
     public EtlResult etl(@PathVariable String type, @PathVariable String task,
                          @RequestParam(name = "params", required = false) String params) {
+        OuterAdapter adapter = loader.getExtension(type);
+        String destination = adapter.getDestination(task);
+        String lockKey = destination == null ? task : destination;
 
-        boolean locked = etlLock.tryLock(ETL_LOCK_ZK_NODE + type + "-" + task);
+        boolean locked = etlLock.tryLock(ETL_LOCK_ZK_NODE + type + "-" + lockKey);
         if (!locked) {
             EtlResult result = new EtlResult();
             result.setSucceeded(false);
@@ -65,13 +68,20 @@ public class CommonRest {
             return result;
         }
         try {
-            OuterAdapter adapter = loader.getExtension(type);
+
-            String destination = adapter.getDestination(task);
+            Boolean oriSwitchStatus;
-            Boolean oriSwithcStatus = null;
             if (destination != null) {
-                oriSwithcStatus = syncSwitch.status(destination);
+                oriSwitchStatus = syncSwitch.status(destination);
+                if (oriSwitchStatus != null && oriSwitchStatus) {
                     syncSwitch.off(destination);
                 }
+            } else {
+                // task可能为destination，直接锁task
+                oriSwitchStatus = syncSwitch.status(task);
+                if (oriSwitchStatus != null && oriSwitchStatus) {
+                    syncSwitch.off(task);
+                }
+            }
             try {
                 List<String> paramArr = null;
                 if (params != null) {
@@ -80,12 +90,14 @@ public class CommonRest {
                 }
                 return adapter.etl(task, paramArr);
             } finally {
-                if (destination != null && oriSwithcStatus != null && oriSwithcStatus) {
+                if (destination != null && oriSwitchStatus != null && oriSwitchStatus) {
                     syncSwitch.on(destination);
+                } else if (destination == null && oriSwitchStatus != null && oriSwitchStatus) {
+                    syncSwitch.on(task);
                 }
             }
         } finally {
-            etlLock.unlock(ETL_LOCK_ZK_NODE + type + "-" + task);
+            etlLock.unlock(ETL_LOCK_ZK_NODE + type + "-" + lockKey);
         }
     }
 

client/pom.xml

@@ -149,6 +149,7 @@
 							<exclude>commons-logging:commons-logging</exclude>
 							<exclude>aopalliance:aopalliance</exclude>
 							<exclude>com.google.protobuf:protobuf-java</exclude>
+							<exclude>com.google.code.findbugs:jsr305</exclude>
 							<exclude>io.netty:*</exclude>
 							<exclude>junit:junit</exclude>
 						</excludes>

server/src/main/java/com/alibaba/otter/canal/kafka/CanalKafkaProducer.java

@@ -105,7 +105,7 @@ public class CanalKafkaProducer implements CanalMQProducer {
                                 canalDestination.getPartition(),
                                 null,
                                 JSON.toJSONString(flatMessage));
-                            producer2.send(record);
+                            producer2.send(record).get();
                         } catch (Exception e) {
                             logger.error(e.getMessage(), e);
                             // producer.abortTransaction();