families) {
+ this.families = families;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) return true;
+ if (o == null || getClass() != o.getClass()) return false;
+
+ HbaseOrm hbaseOrm = (HbaseOrm) o;
+
+ if (table != null ? !table.equals(hbaseOrm.table) : hbaseOrm.table != null) return false;
+ return hbaseTable != null ? hbaseTable.equals(hbaseOrm.hbaseTable) : hbaseOrm.hbaseTable == null;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = table != null ? table.hashCode() : 0;
+ result = 31 * result + (hbaseTable != null ? hbaseTable.hashCode() : 0);
+ return result;
+ }
+ }
+}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java
new file mode 100644
index 00000000..f00e6282
--- /dev/null
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/config/MappingConfigLoader.java
@@ -0,0 +1,149 @@
+package com.alibaba.otter.canal.client.adapter.hbase.config;
+
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.apache.commons.lang.StringUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.yaml.snakeyaml.Yaml;
+
+/**
+ * HBase表映射配置加载器
+ *
+ * 配置统一从hbase-mapping/configs.conf文件作为入口, 该文件包含所有表映射配置的名称或者文件名列表。
+ * 每个对应的表配置可以yml配置文件或者以database.table为配置名的简化形式
+ *
+ *
+ * @author machengyuan 2018-8-21 下午06:45:49
+ * @version 1.0.0
+ */
+public class MappingConfigLoader {
+
+ private static Logger logger = LoggerFactory.getLogger(MappingConfigLoader.class);
+
+ private static final String BASE_PATH = "hbase-mapping/";
+
+ /**
+ * 加载HBase表映射配置
+ *
+ * @return 配置名/配置文件名--对象
+ */
+ public static Map load() {
+ logger.info("## Start loading mapping config ... ");
+ String mappingConfigContent = readConfigContent(BASE_PATH + "configs.conf");
+
+ Map result = new LinkedHashMap<>();
+
+ String[] configLines = mappingConfigContent.split("\n");
+ for (String c : configLines) {
+ if (c == null) {
+ continue;
+ }
+ c = c.trim();
+ if (c.startsWith("#")) {
+ continue;
+ }
+
+ MappingConfig config;
+ String configContent = null;
+
+ if (c.endsWith(".yml")) {
+ configContent = readConfigContent(BASE_PATH + "/" + c);
+ }
+
+ // 简单配置database.table@datasourcekey?rowKey=key1,key2
+ if (StringUtils.isEmpty(configContent)) {
+ String[] mapping = c.split("\\?");
+ String params = mapping.length == 2 ? mapping[1] : null;
+ String rowKey = null;
+ String srcMeta = mapping[0];
+ //
+ if (params != null) {
+ for (String entry : params.split("&")) {
+ if ("rowKey".equals(entry.split("=")[0])) {
+ rowKey = entry.split("=")[1];
+ }
+ }
+ }
+ String dsKey = srcMeta.split("@").length == 2 ? srcMeta.split("@")[1] : null;
+ String[] dbTable;
+ if (dsKey == null) {
+ dbTable = srcMeta.split("\\.");
+ } else {
+ dbTable = srcMeta.split("@")[0].split("\\.");
+ }
+
+ if (dbTable.length == 2) {
+ config = new MappingConfig();
+
+ MappingConfig.HbaseOrm hbaseOrm = new MappingConfig.HbaseOrm();
+ hbaseOrm.setHbaseTable(dbTable[0].toUpperCase() + "." + dbTable[1].toUpperCase());
+ hbaseOrm.setAutoCreateTable(true);
+ hbaseOrm.setDatabase(dbTable[0]);
+ hbaseOrm.setTable(dbTable[1]);
+ hbaseOrm.setMode(MappingConfig.Mode.STRING);
+ hbaseOrm.setRowKey(rowKey);
+ // 有定义rowKey
+ if (rowKey != null) {
+ MappingConfig.ColumnItem columnItem = new MappingConfig.ColumnItem();
+ columnItem.setRowKey(true);
+ columnItem.setColumn(rowKey);
+ hbaseOrm.setRowKeyColumn(columnItem);
+ }
+ config.setHbaseOrm(hbaseOrm);
+
+ } else {
+ throw new RuntimeException(String.format("配置项[%s]内容为空, 或格式不符合database.table", c));
+ }
+
+ } else { // 配置文件配置
+ config = new Yaml().loadAs(configContent, MappingConfig.class);
+ }
+
+ try {
+ config.validate();
+ } catch (Exception e) {
+ throw new RuntimeException("ERROR Config: " + c + " " + e.getMessage(), e);
+ }
+ result.put(c, config);
+ }
+
+ logger.info("## Mapping config loaded");
+ return result;
+ }
+
+ public static String readConfigContent(String config) {
+ InputStream in = null;
+ try {
+ // 先取本地文件，再取类路径
+ File configFile = new File("config/" + config);
+ if (configFile.exists()) {
+ in = new FileInputStream(configFile);
+ } else {
+ in = MappingConfigLoader.class.getClassLoader().getResourceAsStream(config);
+ }
+ if (in == null) {
+ throw new RuntimeException("Config file not found.");
+ }
+
+ byte[] bytes = new byte[in.available()];
+ in.read(bytes);
+ return new String(bytes, "UTF-8");
+ } catch (IOException e) {
+ throw new RuntimeException("Read ds-config.yml or hbase-mappings.conf error. ", e);
+ } finally {
+ try {
+ if (in != null) {
+ in.close();
+ }
+ } catch (IOException e) {
+ // ignore
+ }
+ }
+ }
+}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java
new file mode 100644
index 00000000..13f6a3ac
--- /dev/null
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/service/HbaseSyncService.java
@@ -0,0 +1,418 @@
+package com.alibaba.otter.canal.client.adapter.hbase.service;
+
+import java.util.*;
+
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.otter.canal.client.adapter.hbase.config.MappingConfig;
+import com.alibaba.otter.canal.client.adapter.hbase.support.*;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
+
+/**
+ * HBase同步操作业务
+ *
+ * @author machengyuan 2018-8-21 下午06:45:49
+ * @version 1.0.0
+ */
+public class HbaseSyncService {
+
+ private Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ private HbaseTemplate hbaseTemplate; // HBase操作模板
+
+ public HbaseSyncService(Connection conn){
+ hbaseTemplate = new HbaseTemplate(conn);
+ }
+
+ public void sync(MappingConfig config, Dml dml) {
+ try {
+ if (config != null) {
+ String type = dml.getType();
+ if (type != null && type.equalsIgnoreCase("INSERT")) {
+ insert(config, dml);
+ } else if (type != null && type.equalsIgnoreCase("UPDATE")) {
+ update(config, dml);
+ } else if (type != null && type.equalsIgnoreCase("DELETE")) {
+ delete(config, dml);
+ }
+ if (logger.isDebugEnabled()) {
+ String res = dml.toString();
+ logger.debug(res);
+ }
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+
+ /**
+ * 插入操作
+ *
+ * @param config 配置项
+ * @param dml DML数据
+ */
+ private void insert(MappingConfig config, Dml dml) {
+ List> data = dml.getData();
+ if (data == null || data.isEmpty()) {
+ return;
+ }
+
+ MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
+
+ // if (!validHTable(config)) {
+ // logger.error("HBase table '{}' not exists", hbaseOrm.getHbaseTable());
+ // return;
+ // }
+ int i = 1;
+ boolean complete = false;
+ List rows = new ArrayList<>();
+ for (Map r : data) {
+ HRow hRow = new HRow();
+
+ // 拼接复合rowKey
+ if (hbaseOrm.getRowKey() != null) {
+ String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
+ String rowKeyVale = getRowKeys(rowKeyColumns, r);
+ // params.put("rowKey", Bytes.toBytes(rowKeyVale));
+ hRow.setRowKey(Bytes.toBytes(rowKeyVale));
+ }
+
+ convertData2Row(hbaseOrm, hRow, r);
+ if (hRow.getRowKey() == null) {
+ throw new RuntimeException("empty rowKey");
+ }
+ rows.add(hRow);
+ complete = false;
+ if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rows.isEmpty()) {
+ hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ rows.clear();
+ complete = true;
+ }
+ i++;
+ }
+ if (!complete && !rows.isEmpty()) {
+ hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ }
+
+ }
+
+ /**
+ * 将Map数据转换为HRow行数据
+ *
+ * @param hbaseOrm hbase映射配置
+ * @param hRow 行对象
+ * @param data Map数据
+ */
+ private static void convertData2Row(MappingConfig.HbaseOrm hbaseOrm, HRow hRow, Map data) {
+ Map columnItems = hbaseOrm.getColumnItems();
+ int i = 0;
+ for (Map.Entry entry : data.entrySet()) {
+ if (entry.getValue() != null) {
+ MappingConfig.ColumnItem columnItem = columnItems.get(entry.getKey());
+
+ byte[] bytes = typeConvert(columnItem, hbaseOrm, entry.getValue());
+
+ if (columnItem == null) {
+ String familyName = hbaseOrm.getFamily();
+ String qualifier = entry.getKey();
+ if (hbaseOrm.isUppercaseQualifier()) {
+ qualifier = qualifier.toUpperCase();
+ }
+
+ if (hbaseOrm.getRowKey() == null && i == 0) {
+ hRow.setRowKey(bytes);
+ } else {
+ hRow.addCell(familyName, qualifier, bytes);
+ }
+ } else {
+ if (columnItem.isRowKey()) {
+ // row.put("rowKey", bytes);
+ hRow.setRowKey(bytes);
+ } else {
+ hRow.addCell(columnItem.getFamily(), columnItem.getQualifier(), bytes);
+ }
+ }
+ }
+ i++;
+ }
+ }
+
+ /**
+ * 更新操作
+ *
+ * @param config
+ * @param dml
+ */
+ private void update(MappingConfig config, Dml dml) {
+ List> data = dml.getData();
+ List> old = dml.getOld();
+ if (old == null || old.isEmpty() || data == null || data.isEmpty()) {
+ return;
+ }
+
+ MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
+
+ // if (!validHTable(config)) {
+ // logger.error("HBase table '{}' not exists", hbaseOrm.getHbaseTable());
+ // return;
+ // }
+
+ MappingConfig.ColumnItem rowKeyColumn = hbaseOrm.getRowKeyColumn();
+ int index = 0;
+ int i = 1;
+ boolean complete = false;
+ List rows = new ArrayList<>();
+ out: for (Map r : data) {
+ byte[] rowKeyBytes;
+
+ if (hbaseOrm.getRowKey() != null) {
+ String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
+
+ // 判断是否有复合主键修改
+ for (String updateColumn : old.get(index).keySet()) {
+ for (String rowKeyColumnName : rowKeyColumns) {
+ if (rowKeyColumnName.equalsIgnoreCase(updateColumn)) {
+ // 调用删除插入操作
+ deleteAndInsert(config, dml);
+ continue out;
+ }
+ }
+ }
+
+ String rowKeyVale = getRowKeys(rowKeyColumns, r);
+ rowKeyBytes = Bytes.toBytes(rowKeyVale);
+ } else if (rowKeyColumn == null) {
+ Map rowKey = data.get(0);
+ rowKeyBytes = typeConvert(null, hbaseOrm, rowKey.values().iterator().next());
+ } else {
+ rowKeyBytes = typeConvert(rowKeyColumn, hbaseOrm, r.get(rowKeyColumn.getColumn()));
+ }
+ if (rowKeyBytes == null) throw new RuntimeException("rowKey值为空");
+
+ Map columnItems = hbaseOrm.getColumnItems();
+ HRow hRow = new HRow(rowKeyBytes);
+ for (String updateColumn : old.get(index).keySet()) {
+ MappingConfig.ColumnItem columnItem = columnItems.get(updateColumn);
+ if (columnItem == null) {
+ String family = hbaseOrm.getFamily();
+ String qualifier = updateColumn;
+ if (hbaseOrm.isUppercaseQualifier()) {
+ qualifier = qualifier.toUpperCase();
+ }
+
+ Object newVal = r.get(updateColumn);
+
+ if (newVal == null) {
+ hRow.addCell(family, qualifier, null);
+ } else {
+ hRow.addCell(family, qualifier, typeConvert(null, hbaseOrm, newVal));
+ }
+ } else {
+ // 排除修改id的情况
+ if (columnItem.isRowKey()) continue;
+
+ Object newVal = r.get(updateColumn);
+ if (newVal == null) {
+ hRow.addCell(columnItem.getFamily(), columnItem.getQualifier(), null);
+ } else {
+ hRow.addCell(columnItem.getFamily(),
+ columnItem.getQualifier(),
+ typeConvert(columnItem, hbaseOrm, newVal));
+ }
+ }
+ }
+ rows.add(hRow);
+ complete = false;
+ if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rows.isEmpty()) {
+ hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ rows.clear();
+ complete = true;
+ }
+ i++;
+ index++;
+ }
+ if (!complete && !rows.isEmpty()) {
+ hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ }
+ }
+
+ private void delete(MappingConfig config, Dml dml) {
+ List> data = dml.getData();
+ if (data == null || data.isEmpty()) {
+ return;
+ }
+
+ MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
+
+ // if (!validHTable(config)) {
+ // logger.error("HBase table '{}' not exists", hbaseOrm.getHbaseTable());
+ // return;
+ // }
+
+ MappingConfig.ColumnItem rowKeyColumn = hbaseOrm.getRowKeyColumn();
+ boolean complete = false;
+ int i = 1;
+ Set rowKeys = new HashSet<>();
+ for (Map r : data) {
+ byte[] rowKeyBytes;
+
+ if (hbaseOrm.getRowKey() != null) {
+ String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
+ String rowKeyVale = getRowKeys(rowKeyColumns, r);
+ rowKeyBytes = Bytes.toBytes(rowKeyVale);
+ } else if (rowKeyColumn == null) {
+ // 如果不需要类型转换
+ Map rowKey = data.get(0);
+ rowKeyBytes = typeConvert(null, hbaseOrm, rowKey.values().iterator().next());
+ } else {
+ Object val = r.get(rowKeyColumn.getColumn());
+ rowKeyBytes = typeConvert(rowKeyColumn, hbaseOrm, val);
+ }
+ if (rowKeyBytes == null) throw new RuntimeException("rowKey值为空");
+ rowKeys.add(rowKeyBytes);
+ complete = false;
+ if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rowKeys.isEmpty()) {
+ hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
+ rowKeys.clear();
+ complete = true;
+ }
+ i++;
+ }
+ if (!complete && !rowKeys.isEmpty()) {
+ hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
+ }
+ }
+
+ private void deleteAndInsert(MappingConfig config, Dml dml) {
+ List> data = dml.getData();
+ List> old = dml.getOld();
+ if (old == null || old.isEmpty() || data == null || data.isEmpty()) {
+ return;
+ }
+ MappingConfig.HbaseOrm hbaseOrm = config.getHbaseOrm();
+
+ String[] rowKeyColumns = hbaseOrm.getRowKey().trim().split(",");
+
+ int index = 0;
+ int i = 1;
+ boolean complete = false;
+ Set rowKeys = new HashSet<>();
+ List rows = new ArrayList<>();
+ for (Map r : data) {
+ // 拼接老的rowKey
+ List updateSubRowKey = new ArrayList<>();
+ for (String rowKeyColumnName : rowKeyColumns) {
+ for (String updateColumn : old.get(index).keySet()) {
+ if (rowKeyColumnName.equalsIgnoreCase(updateColumn)) {
+ updateSubRowKey.add(rowKeyColumnName);
+ }
+ }
+ }
+ if (updateSubRowKey.isEmpty()) {
+ throw new RuntimeException("没有更新复合主键的RowKey");
+ }
+ StringBuilder oldRowKey = new StringBuilder();
+ StringBuilder newRowKey = new StringBuilder();
+ for (String rowKeyColumnName : rowKeyColumns) {
+ newRowKey.append(r.get(rowKeyColumnName).toString()).append("|");
+ if (!updateSubRowKey.contains(rowKeyColumnName)) {
+ // 从data取
+ oldRowKey.append(r.get(rowKeyColumnName).toString()).append("|");
+ } else {
+ // 从old取
+ oldRowKey.append(old.get(index).get(rowKeyColumnName).toString()).append("|");
+ }
+ }
+ int len = newRowKey.length();
+ newRowKey.delete(len - 1, len);
+ len = oldRowKey.length();
+ oldRowKey.delete(len - 1, len);
+ byte[] newRowKeyBytes = Bytes.toBytes(newRowKey.toString());
+ byte[] oldRowKeyBytes = Bytes.toBytes(oldRowKey.toString());
+
+ rowKeys.add(oldRowKeyBytes);
+ HRow row = new HRow(newRowKeyBytes);
+ convertData2Row(hbaseOrm, row, r);
+ rows.add(row);
+ complete = false;
+ if (i % config.getHbaseOrm().getCommitBatch() == 0 && !rows.isEmpty()) {
+ hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
+
+ hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ rowKeys.clear();
+ rows.clear();
+ complete = true;
+ }
+ i++;
+ index++;
+ }
+ if (!complete && !rows.isEmpty()) {
+ hbaseTemplate.deletes(hbaseOrm.getHbaseTable(), rowKeys);
+ hbaseTemplate.puts(hbaseOrm.getHbaseTable(), rows);
+ }
+ }
+
+ /**
+ * 根据对应的类型进行转换
+ *
+ * @param columnItem 列项配置
+ * @param hbaseOrm hbase映射配置
+ * @param value 值
+ * @return 复合字段rowKey
+ */
+ private static byte[] typeConvert(MappingConfig.ColumnItem columnItem, MappingConfig.HbaseOrm hbaseOrm,
+ Object value) {
+ if (value == null) {
+ return null;
+ }
+ byte[] bytes = null;
+ if (columnItem == null || columnItem.getType() == null || "".equals(columnItem.getType())) {
+ if (MappingConfig.Mode.STRING == hbaseOrm.getMode()) {
+ bytes = Bytes.toBytes(value.toString());
+ } else if (MappingConfig.Mode.NATIVE == hbaseOrm.getMode()) {
+ bytes = TypeUtil.toBytes(value);
+ } else if (MappingConfig.Mode.PHOENIX == hbaseOrm.getMode()) {
+ PhType phType = PhType.getType(value.getClass());
+ bytes = PhTypeUtil.toBytes(value, phType);
+ }
+ } else {
+ if (hbaseOrm.getMode() == MappingConfig.Mode.STRING) {
+ bytes = Bytes.toBytes(value.toString());
+ } else if (hbaseOrm.getMode() == MappingConfig.Mode.NATIVE) {
+ Type type = Type.getType(columnItem.getType());
+ bytes = TypeUtil.toBytes(value, type);
+ } else if (hbaseOrm.getMode() == MappingConfig.Mode.PHOENIX) {
+ PhType phType = PhType.getType(columnItem.getType());
+ bytes = PhTypeUtil.toBytes(value, phType);
+ }
+ }
+ return bytes;
+ }
+
+ /**
+ * 获取复合字段作为rowKey的拼接
+ *
+ * @param rowKeyColumns 复合rowK对应的字段
+ * @param data 数据
+ * @return
+ */
+ private static String getRowKeys(String[] rowKeyColumns, Map data) {
+ StringBuilder rowKeyValue = new StringBuilder();
+ for (String rowKeyColumnName : rowKeyColumns) {
+ Object obj = data.get(rowKeyColumnName);
+ if (obj != null) {
+ rowKeyValue.append(obj.toString());
+ }
+ rowKeyValue.append("|");
+ }
+ int len = rowKeyValue.length();
+ if (len > 0) {
+ rowKeyValue.delete(len - 1, len);
+ }
+ return rowKeyValue.toString();
+ }
+
+}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HRow.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HRow.java
new file mode 100644
index 00000000..adeb2ff5
--- /dev/null
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HRow.java
@@ -0,0 +1,84 @@
+package com.alibaba.otter.canal.client.adapter.hbase.support;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * HBase操作对象类
+ *
+ * @author machengyuan 2018-8-21 下午10:12:34
+ * @version 1.0.0
+ */
+public class HRow {
+
+ private byte[] rowKey;
+ private List cells = new ArrayList<>();
+
+ public HRow(){
+ }
+
+ public HRow(byte[] rowKey){
+ this.rowKey = rowKey;
+ }
+
+ public byte[] getRowKey() {
+ return rowKey;
+ }
+
+ public void setRowKey(byte[] rowKey) {
+ this.rowKey = rowKey;
+ }
+
+ public List getCells() {
+ return cells;
+ }
+
+ public void setCells(List cells) {
+ this.cells = cells;
+ }
+
+ public void addCell(String family, String qualifier, byte[] value) {
+ HCell hCell = new HCell(family, qualifier, value);
+ cells.add(hCell);
+ }
+
+ public static class HCell {
+
+ private String family;
+ private String qualifier;
+ private byte[] value;
+
+ public HCell(){
+ }
+
+ public HCell(String family, String qualifier, byte[] value){
+ this.family = family;
+ this.qualifier = qualifier;
+ this.value = value;
+ }
+
+ public String getFamily() {
+ return family;
+ }
+
+ public void setFamily(String family) {
+ this.family = family;
+ }
+
+ public String getQualifier() {
+ return qualifier;
+ }
+
+ public void setQualifier(String qualifier) {
+ this.qualifier = qualifier;
+ }
+
+ public byte[] getValue() {
+ return value;
+ }
+
+ public void setValue(byte[] value) {
+ this.value = value;
+ }
+ }
+}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HbaseTemplate.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HbaseTemplate.java
new file mode 100644
index 00000000..62b7ca03
--- /dev/null
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HbaseTemplate.java
@@ -0,0 +1,159 @@
+package com.alibaba.otter.canal.client.adapter.hbase.support;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+
+import org.apache.hadoop.hbase.HColumnDescriptor;
+import org.apache.hadoop.hbase.HTableDescriptor;
+import org.apache.hadoop.hbase.TableName;
+import org.apache.hadoop.hbase.client.*;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * HBase操作模板
+ *
+ * @author machengyuan 2018-8-21 下午10:12:34
+ * @version 1.0.0
+ */
+public class HbaseTemplate {
+
+ private Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ private Connection conn;
+
+ public HbaseTemplate(Connection conn){
+ this.conn = conn;
+ }
+
+ public boolean tableExists(String tableName) {
+ try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
+
+ return admin.tableExists(TableName.valueOf(tableName));
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void createTable(String tableName, String... familyNames) {
+ try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
+
+ HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
+ // 添加列簇
+ if (familyNames != null) {
+ for (String familyName : familyNames) {
+ HColumnDescriptor hcd = new HColumnDescriptor(familyName);
+ desc.addFamily(hcd);
+ }
+ }
+ admin.createTable(desc);
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void disableTable(String tableName) {
+ try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
+ admin.disableTable(tableName);
+ } catch (IOException e) {
+ logger.error(e.getMessage(), e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ public void deleteTable(String tableName) {
+ try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
+ if (admin.isTableEnabled(tableName)) {
+ disableTable(tableName);
+ }
+ admin.deleteTable(tableName);
+ } catch (IOException e) {
+ logger.error(e.getMessage(), e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * 插入一行数据
+ *
+ * @param tableName 表名
+ * @param hRow 行数据对象
+ * @return 是否成功
+ */
+ public Boolean put(String tableName, HRow hRow) {
+ boolean flag = false;
+ try {
+ HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
+ Put put = new Put(hRow.getRowKey());
+ for (HRow.HCell hCell : hRow.getCells()) {
+ put.addColumn(Bytes.toBytes(hCell.getFamily()), Bytes.toBytes(hCell.getQualifier()), hCell.getValue());
+ }
+ table.put(put);
+ flag = true;
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ return flag;
+
+ }
+
+ /**
+ * 批量插入
+ *
+ * @param tableName 表名
+ * @param rows 行数据对象集合
+ * @return 是否成功
+ */
+ public Boolean puts(String tableName, List rows) {
+ boolean flag = false;
+ try {
+ HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
+ List puts = new ArrayList<>();
+ for (HRow hRow : rows) {
+ Put put = new Put(hRow.getRowKey());
+ for (HRow.HCell hCell : hRow.getCells()) {
+ put.addColumn(Bytes.toBytes(hCell.getFamily()),
+ Bytes.toBytes(hCell.getQualifier()),
+ hCell.getValue());
+ }
+ puts.add(put);
+ }
+ if (!puts.isEmpty()) {
+ table.put(puts);
+ }
+ flag = true;
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ return flag;
+ }
+
+ /**
+ * 批量删除数据
+ *
+ * @param tableName 表名
+ * @param rowKeys rowKey集合
+ * @return 是否成功
+ */
+ public Boolean deletes(String tableName, Set rowKeys) {
+ boolean flag = false;
+ try {
+ HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
+ List deletes = new ArrayList<>();
+ for (byte[] rowKey : rowKeys) {
+ Delete delete = new Delete(rowKey);
+ deletes.add(delete);
+ }
+ if (!deletes.isEmpty()) {
+ table.delete(deletes);
+ }
+ flag = true;
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ }
+ return flag;
+ }
+}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhType.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhType.java
new file mode 100644
index 00000000..09c63a6c
--- /dev/null
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhType.java
@@ -0,0 +1,150 @@
+package com.alibaba.otter.canal.client.adapter.hbase.support;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Date;
+
+/**
+ * Phoenix类型
+ *
+ * @author machengyuan 2018-8-21 下午06:12:34
+ * @version 1.0.0
+ */
+public enum PhType {
+ DEFAULT(-1, "VARCHAR"),
+ UNSIGNED_INT(4, "UNSIGNED_INT"),
+ UNSIGNED_LONG(8, "UNSIGNED_LONG"),
+ UNSIGNED_TINYINT(1, "UNSIGNED_TINYINT"),
+ UNSIGNED_SMALLINT(2, "UNSIGNED_SMALLINT"),
+ UNSIGNED_FLOAT(4, "UNSIGNED_FLOAT"),
+ UNSIGNED_DOUBLE(8, "UNSIGNED_DOUBLE"),
+ INTEGER(4, "INTEGER"),
+ BIGINT(8, "BIGINT"),
+ TINYINT(1, "TINYINT"),
+ SMALLINT(2, "SMALLINT"),
+ FLOAT(4, "FLOAT"),
+ DOUBLE(8, "DOUBLE"),
+ DECIMAL(-1, "DECIMAL"),
+ BOOLEAN(1, "BOOLEAN"),
+ UNSIGNED_TIME(8, "UNSIGNED_TIME"),
+ UNSIGNED_DATE(8, "UNSIGNED_DATE"),
+ UNSIGNED_TIMESTAMP(12, "UNSIGNED_TIMESTAMP"),
+ TIME(8, "TIME"),
+ DATE(8, "DATE"),
+ TIMESTAMP(12, "TIMESTAMP"),
+ VARCHAR(-1, "VARCHAR"),
+ VARBINARY(-1, "VARBINARY");
+
+ /**
+ * -1：长度可变
+ */
+ private int len;
+ private String type;
+
+ PhType(int len, String type) {
+ this.len = len;
+ this.type = type;
+ }
+
+ public int getLen() {
+ return len;
+ }
+
+ public String getType() {
+ return this.type;
+ }
+
+ public static PhType getType(Class javaType) {
+ if (javaType == null) return DEFAULT;
+ PhType phType;
+ if (Integer.class.isAssignableFrom(javaType) || int.class.isAssignableFrom(javaType)) {
+ phType = INTEGER;
+ } else if (Long.class.isAssignableFrom(javaType) || long.class.isAssignableFrom(javaType)) {
+ phType = BIGINT;
+ } else if (Byte.class.isAssignableFrom(javaType) || byte.class.isAssignableFrom(javaType)) {
+ phType = TINYINT;
+ } else if (Short.class.isAssignableFrom(javaType) || short.class.isAssignableFrom(javaType)) {
+ phType = SMALLINT;
+ } else if (Float.class.isAssignableFrom(javaType) || float.class.isAssignableFrom(javaType)) {
+ phType = FLOAT;
+ } else if (Double.class.isAssignableFrom(javaType) || double.class.isAssignableFrom(javaType)) {
+ phType = DOUBLE;
+ } else if (Boolean.class.isAssignableFrom(javaType) || boolean.class.isAssignableFrom(javaType)) {
+ phType = BOOLEAN;
+ } else if (java.sql.Date.class.isAssignableFrom(javaType)) {
+ phType = DATE;
+ } else if (Time.class.isAssignableFrom(javaType)) {
+ phType = DATE;
+ } else if (Timestamp.class.isAssignableFrom(javaType)) {
+ phType = TIMESTAMP;
+ } else if (Date.class.isAssignableFrom(javaType)) {
+ phType = DATE;
+ } else if (byte[].class.isAssignableFrom(javaType)) {
+ phType = VARBINARY;
+ } else if (String.class.isAssignableFrom(javaType)) {
+ phType = VARCHAR;
+ } else if (BigDecimal.class.isAssignableFrom(javaType)) {
+ phType = DECIMAL;
+ } else if (BigInteger.class.isAssignableFrom(javaType)) {
+ phType = UNSIGNED_LONG;
+ } else {
+ phType = DEFAULT;
+ }
+ return phType;
+ }
+
+ public static PhType getType(String type) {
+ if (type == null) return DEFAULT;
+ PhType phType;
+ if (type.equalsIgnoreCase(UNSIGNED_INT.type)) {
+ phType = UNSIGNED_INT;
+ } else if (type.equalsIgnoreCase(UNSIGNED_LONG.type)) {
+ phType = UNSIGNED_LONG;
+ } else if (type.equalsIgnoreCase(UNSIGNED_TINYINT.type)) {
+ phType = UNSIGNED_TINYINT;
+ } else if (type.equalsIgnoreCase(UNSIGNED_SMALLINT.type)) {
+ phType = UNSIGNED_SMALLINT;
+ } else if (type.equalsIgnoreCase(UNSIGNED_FLOAT.type)) {
+ phType = UNSIGNED_FLOAT;
+ } else if (type.equalsIgnoreCase(UNSIGNED_DOUBLE.type)) {
+ phType = UNSIGNED_DOUBLE;
+ } else if (type.equalsIgnoreCase(INTEGER.type)) {
+ phType = INTEGER;
+ } else if (type.equalsIgnoreCase(BIGINT.type)) {
+ phType = BIGINT;
+ } else if (type.equalsIgnoreCase(TINYINT.type)) {
+ phType = TINYINT;
+ } else if (type.equalsIgnoreCase(SMALLINT.type)) {
+ phType = SMALLINT;
+ } else if (type.equalsIgnoreCase(FLOAT.type)) {
+ phType = FLOAT;
+ } else if (type.equalsIgnoreCase(DOUBLE.type)) {
+ phType = DOUBLE;
+ } else if (type.equalsIgnoreCase(BOOLEAN.type)) {
+ phType = BOOLEAN;
+ } else if (type.equalsIgnoreCase(UNSIGNED_TIME.type)) {
+ phType = UNSIGNED_TIME;
+ } else if (type.equalsIgnoreCase(UNSIGNED_DATE.type)) {
+ phType = UNSIGNED_DATE;
+ } else if (type.equalsIgnoreCase(UNSIGNED_TIMESTAMP.type)) {
+ phType = UNSIGNED_TIMESTAMP;
+ } else if (type.equalsIgnoreCase(TIME.type)) {
+ phType = TIME;
+ } else if (type.equalsIgnoreCase(DATE.type)) {
+ phType = DATE;
+ } else if (type.equalsIgnoreCase(TIMESTAMP.type)) {
+ phType = TIMESTAMP;
+ } else if (type.equalsIgnoreCase(VARCHAR.type)) {
+ phType = VARCHAR;
+ } else if (type.equalsIgnoreCase(VARBINARY.type)) {
+ phType = VARBINARY;
+ } else if (type.equalsIgnoreCase(DECIMAL.type)) {
+ phType = DECIMAL;
+ } else {
+ phType = DEFAULT;
+ }
+ return phType;
+ }
+}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java
new file mode 100644
index 00000000..25473681
--- /dev/null
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/PhTypeUtil.java
@@ -0,0 +1,614 @@
+package com.alibaba.otter.canal.client.adapter.hbase.support;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.math.MathContext;
+import java.math.RoundingMode;
+import java.sql.Timestamp;
+import java.util.Date;
+
+import org.apache.hadoop.hbase.util.Bytes;
+import org.joda.time.DateTime;
+
+import com.google.common.math.LongMath;
+
+/**
+ * Phoenix类型转换工具类
+ *
+ * @author machengyuan 2018-8-21 下午06:14:26
+ * @version 1.0.0
+ */
+public class PhTypeUtil {
+
+ public static byte[] toBytes(Object v, PhType phType) {
+ if (v == null) return null;
+ byte[] b = null;
+ if (phType == PhType.DEFAULT) {
+ PhType phType1 = PhType.getType(v.getClass());
+ if (phType1 != null && phType1 != PhType.DEFAULT) {
+ toBytes(v, phType1);
+ }
+ } else if (phType == PhType.INTEGER) {
+ b = new byte[Bytes.SIZEOF_INT];
+ encodeInt(((Number) v).intValue(), b, 0);
+ } else if (phType == PhType.UNSIGNED_INT) {
+ b = new byte[Bytes.SIZEOF_INT];
+ encodeUnsignedInt(((Number) v).intValue(), b, 0);
+ } else if (phType == PhType.BIGINT) {
+ b = new byte[Bytes.SIZEOF_LONG];
+ encodeLong(((Number) v).longValue(), b, 0);
+ } else if (phType == PhType.UNSIGNED_LONG) {
+ b = new byte[Bytes.SIZEOF_LONG];
+ encodeUnsignedLong(((Number) v).longValue(), b, 0);
+ } else if (phType == PhType.SMALLINT) {
+ b = new byte[Bytes.SIZEOF_SHORT];
+ encodeShort(((Number) v).shortValue(), b, 0);
+ } else if (phType == PhType.UNSIGNED_SMALLINT) {
+ b = new byte[Bytes.SIZEOF_SHORT];
+ encodeUnsignedShort(((Number) v).shortValue(), b, 0);
+ } else if (phType == PhType.TINYINT) {
+ b = new byte[Bytes.SIZEOF_BYTE];
+ encodeByte(((Number) v).byteValue(), b, 0);
+ } else if (phType == PhType.UNSIGNED_TINYINT) {
+ b = new byte[Bytes.SIZEOF_BYTE];
+ encodeUnsignedByte(((Number) v).byteValue(), b, 0);
+ } else if (phType == PhType.FLOAT) {
+ b = new byte[Bytes.SIZEOF_FLOAT];
+ encodeFloat(((Number) v).floatValue(), b, 0);
+ } else if (phType == PhType.UNSIGNED_FLOAT) {
+ b = new byte[Bytes.SIZEOF_FLOAT];
+ encodeUnsignedFloat(((Number) v).floatValue(), b, 0);
+ } else if (phType == PhType.DOUBLE) {
+ b = new byte[Bytes.SIZEOF_DOUBLE];
+ encodeDouble(((Number) v).doubleValue(), b, 0);
+ } else if (phType == PhType.UNSIGNED_DOUBLE) {
+ b = new byte[Bytes.SIZEOF_DOUBLE];
+ encodeUnsignedDouble(((Number) v).doubleValue(), b, 0);
+ } else if (phType == PhType.BOOLEAN) {
+ if ((Boolean) v) {
+ b = new byte[] { 1 };
+ } else {
+ b = new byte[] { 0 };
+ }
+ } else if (phType == PhType.TIME || phType == PhType.DATE) {
+ b = new byte[Bytes.SIZEOF_LONG];
+ encodeDate(v, b, 0);
+ } else if (phType == PhType.TIMESTAMP) {
+ b = new byte[Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT];
+ encodeTimestamp(v, b, 0);
+ } else if (phType == PhType.UNSIGNED_TIME || phType == PhType.UNSIGNED_DATE) {
+ b = new byte[Bytes.SIZEOF_LONG];
+ encodeUnsignedDate(v, b, 0);
+ } else if (phType == PhType.UNSIGNED_TIMESTAMP) {
+ b = new byte[Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT];
+ encodeUnsignedTimestamp(v, b, 0);
+ } else if (phType == PhType.VARBINARY) {
+ b = (byte[]) v;
+ } else if (phType == PhType.VARCHAR) {
+ b = Bytes.toBytes(v.toString());
+ } else if (phType == PhType.DECIMAL) {
+ if (v instanceof BigDecimal) {
+ b = encodeDecimal(v);
+ } else if (v instanceof Number) {
+ b = encodeDecimal(new BigDecimal(v.toString()));
+ }
+ }
+ return b;
+ }
+
+ public static Object toObject(byte[] b, PhType phType) {
+ if (b == null) return null;
+ Object v = null;
+ if (phType == PhType.INTEGER) {
+ v = decodeInt(b, 0);
+ } else if (phType == PhType.UNSIGNED_INT) {
+ v = decodeUnsignedInt(b, 0);
+ } else if (phType == PhType.BIGINT) {
+ v = decodeLong(b, 0);
+ } else if (phType == PhType.UNSIGNED_LONG) {
+ v = decodeUnsignedLong(b, 0);
+ } else if (phType == PhType.SMALLINT) {
+ v = decodeShort(b, 0);
+ } else if (phType == PhType.UNSIGNED_SMALLINT) {
+ v = decodeUnsignedShort(b, 0);
+ } else if (phType == PhType.TINYINT) {
+ v = decodeByte(b, 0);
+ } else if (phType == PhType.UNSIGNED_TINYINT) {
+ v = decodeUnsignedByte(b, 0);
+ } else if (phType == PhType.FLOAT) {
+ v = decodeFloat(b, 0);
+ } else if (phType == PhType.UNSIGNED_FLOAT) {
+ v = decodeUnsignedFloat(b, 0);
+ } else if (phType == PhType.DOUBLE) {
+ v = decodeDouble(b, 0);
+ } else if (phType == PhType.UNSIGNED_DOUBLE) {
+ v = decodeUnsignedDouble(b, 0);
+ } else if (phType == PhType.BOOLEAN) {
+ checkForSufficientLength(b, 0, Bytes.SIZEOF_BOOLEAN);
+ if (b[0] == 1) {
+ v = true;
+ } else if (b[0] == 0) {
+ v = false;
+ }
+ } else if (phType == PhType.TIME || phType == PhType.DATE) {
+ v = new Date(decodeLong(b, 0));
+ } else if (phType == PhType.TIMESTAMP) {
+ long millisDeserialized = decodeLong(b, 0);
+ Timestamp ts = new Timestamp(millisDeserialized);
+ int nanosDeserialized = decodeUnsignedInt(b, Bytes.SIZEOF_LONG);
+ ts.setNanos(nanosDeserialized < 1000000 ? ts.getNanos() + nanosDeserialized : nanosDeserialized);
+ v = ts;
+ } else if (phType == PhType.UNSIGNED_TIME || phType == PhType.UNSIGNED_DATE) {
+ v = new Date(decodeUnsignedLong(b, 0));
+ } else if (phType == PhType.UNSIGNED_TIMESTAMP) {
+ long millisDeserialized = decodeUnsignedLong(b, 0);
+ Timestamp ts = new Timestamp(millisDeserialized);
+ int nanosDeserialized = decodeUnsignedInt(b, Bytes.SIZEOF_LONG);
+ ts.setNanos(nanosDeserialized < 1000000 ? ts.getNanos() + nanosDeserialized : nanosDeserialized);
+ v = ts;
+ } else if (phType == PhType.VARBINARY) {
+ v = b;
+ } else if (phType == PhType.VARCHAR || phType == PhType.DEFAULT) {
+ v = Bytes.toString(b);
+ } else if (phType == PhType.DECIMAL) {
+ v = decodeDecimal(b, 0, b.length);
+ }
+
+ return v;
+ }
+
+ private static int decodeInt(byte[] bytes, int o) {
+ checkForSufficientLength(bytes, o, Bytes.SIZEOF_INT);
+ int v;
+ v = bytes[o] ^ 0x80; // Flip sign bit back
+ for (int i = 1; i < Bytes.SIZEOF_INT; i++) {
+ v = (v << 8) + (bytes[o + i] & 0xff);
+ }
+ return v;
+ }
+
+ private static int encodeInt(int v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_INT);
+ b[o + 0] = (byte) ((v >> 24) ^ 0x80); // Flip sign bit so that INTEGER is binary comparable
+ b[o + 1] = (byte) (v >> 16);
+ b[o + 2] = (byte) (v >> 8);
+ b[o + 3] = (byte) v;
+ return Bytes.SIZEOF_INT;
+ }
+
+ private static int decodeUnsignedInt(byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_INT);
+
+ int v = Bytes.toInt(b, o);
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ return v;
+ }
+
+ private static int encodeUnsignedInt(int v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_INT);
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ Bytes.putInt(b, o, v);
+ return Bytes.SIZEOF_INT;
+ }
+
+ private static long decodeLong(byte[] bytes, int o) {
+ checkForSufficientLength(bytes, o, Bytes.SIZEOF_LONG);
+ long v;
+ byte b = bytes[o];
+ v = b ^ 0x80; // Flip sign bit back
+ for (int i = 1; i < Bytes.SIZEOF_LONG; i++) {
+ b = bytes[o + i];
+ v = (v << 8) + (b & 0xff);
+ }
+ return v;
+ }
+
+ private static int encodeLong(long v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_LONG);
+ b[o + 0] = (byte) ((v >> 56) ^ 0x80); // Flip sign bit so that INTEGER is binary comparable
+ b[o + 1] = (byte) (v >> 48);
+ b[o + 2] = (byte) (v >> 40);
+ b[o + 3] = (byte) (v >> 32);
+ b[o + 4] = (byte) (v >> 24);
+ b[o + 5] = (byte) (v >> 16);
+ b[o + 6] = (byte) (v >> 8);
+ b[o + 7] = (byte) v;
+ return Bytes.SIZEOF_LONG;
+ }
+
+ private static long decodeUnsignedLong(byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_LONG);
+ long v = 0;
+ for (int i = o; i < o + Bytes.SIZEOF_LONG; i++) {
+ v <<= 8;
+ v ^= b[i] & 0xFF;
+ }
+
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ return v;
+ }
+
+ private static int encodeUnsignedLong(long v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_LONG);
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ Bytes.putLong(b, o, v);
+ return Bytes.SIZEOF_LONG;
+ }
+
+ private static short decodeShort(byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT);
+ int v;
+ v = b[o] ^ 0x80; // Flip sign bit back
+ for (int i = 1; i < Bytes.SIZEOF_SHORT; i++) {
+ v = (v << 8) + (b[o + i] & 0xff);
+ }
+ return (short) v;
+ }
+
+ private static int encodeShort(short v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT);
+ b[o + 0] = (byte) ((v >> 8) ^ 0x80); // Flip sign bit so that Short is binary comparable
+ b[o + 1] = (byte) v;
+ return Bytes.SIZEOF_SHORT;
+ }
+
+ private static short decodeUnsignedShort(byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT);
+ short v = Bytes.toShort(b, o);
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ return v;
+ }
+
+ private static int encodeUnsignedShort(short v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_SHORT);
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ Bytes.putShort(b, o, v);
+ return Bytes.SIZEOF_SHORT;
+ }
+
+ private static byte decodeByte(byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_BYTE);
+ int v;
+ v = b[o] ^ 0x80; // Flip sign bit back
+ return (byte) v;
+ }
+
+ private static int encodeByte(byte v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_BYTE);
+ b[o] = (byte) (v ^ 0x80); // Flip sign bit so that Short is binary comparable
+ return Bytes.SIZEOF_BYTE;
+ }
+
+ private static byte decodeUnsignedByte(byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_BYTE);
+ byte v = b[o];
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ return v;
+ }
+
+ private static int encodeUnsignedByte(byte v, byte[] b, int o) {
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ Bytes.putByte(b, o, v);
+ return Bytes.SIZEOF_BYTE;
+ }
+
+ private static float decodeFloat(byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_INT);
+ int value;
+ value = Bytes.toInt(b, o);
+ value--;
+ value ^= (~value >> Integer.SIZE - 1) | Integer.MIN_VALUE;
+ return Float.intBitsToFloat(value);
+ }
+
+ private static int encodeFloat(float v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_FLOAT);
+ int i = Float.floatToIntBits(v);
+ i = (i ^ ((i >> Integer.SIZE - 1) | Integer.MIN_VALUE)) + 1;
+ Bytes.putInt(b, o, i);
+ return Bytes.SIZEOF_FLOAT;
+ }
+
+ private static float decodeUnsignedFloat(byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_FLOAT);
+ float v = Bytes.toFloat(b, o);
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ return v;
+ }
+
+ private static int encodeUnsignedFloat(float v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_FLOAT);
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ Bytes.putFloat(b, o, v);
+ return Bytes.SIZEOF_FLOAT;
+ }
+
+ private static double decodeDouble(byte[] bytes, int o) {
+ checkForSufficientLength(bytes, o, Bytes.SIZEOF_LONG);
+ long l;
+ l = Bytes.toLong(bytes, o);
+ l--;
+ l ^= (~l >> Long.SIZE - 1) | Long.MIN_VALUE;
+ return Double.longBitsToDouble(l);
+ }
+
+ private static int encodeDouble(double v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_LONG);
+ long l = Double.doubleToLongBits(v);
+ l = (l ^ ((l >> Long.SIZE - 1) | Long.MIN_VALUE)) + 1;
+ Bytes.putLong(b, o, l);
+ return Bytes.SIZEOF_LONG;
+ }
+
+ private static double decodeUnsignedDouble(byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_DOUBLE);
+ double v = Bytes.toDouble(b, o);
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ return v;
+ }
+
+ private static int encodeUnsignedDouble(double v, byte[] b, int o) {
+ checkForSufficientLength(b, o, Bytes.SIZEOF_DOUBLE);
+ if (v < 0) {
+ throw new RuntimeException();
+ }
+ Bytes.putDouble(b, o, v);
+ return Bytes.SIZEOF_DOUBLE;
+ }
+
+ private static int encodeDate(Object v, byte[] b, int o) {
+ if (v instanceof Date) {
+ encodeLong(((Date) v).getTime(), b, 0);
+ } else if (v instanceof String) {
+ String dateStr = (String) v;
+ Date date;
+ try {
+ date = parseDatetime(dateStr);
+ if (date != null) {
+ encodeLong(date.getTime(), b, 0);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return Bytes.SIZEOF_LONG;
+ }
+
+ private static int encodeTimestamp(Object v, byte[] b, int o) {
+ if (v instanceof Timestamp) {
+ Timestamp ts = (Timestamp) v;
+ encodeLong(ts.getTime(), b, o);
+ Bytes.putInt(b, Bytes.SIZEOF_LONG, ts.getNanos() % 1000000);
+ } else {
+ encodeDate(v, b, o);
+ }
+ return Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT;
+ }
+
+ private static int encodeUnsignedDate(Object v, byte[] b, int o) {
+ if (v instanceof Date) {
+ encodeUnsignedLong(((Date) v).getTime(), b, 0);
+ } else if (v instanceof String) {
+ String dateStr = (String) v;
+ Date date;
+ try {
+ date = parseDatetime(dateStr);
+ if (date != null) {
+ encodeUnsignedLong(date.getTime(), b, 0);
+ }
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ }
+ return Bytes.SIZEOF_LONG;
+ }
+
+ private static int encodeUnsignedTimestamp(Object v, byte[] b, int o) {
+ if (v instanceof Timestamp) {
+ Timestamp ts = (Timestamp) v;
+ encodeUnsignedLong(ts.getTime(), b, o);
+ Bytes.putInt(b, Bytes.SIZEOF_LONG, ts.getNanos() % 1000000);
+ } else {
+ encodeUnsignedDate(v, b, o);
+ }
+ return Bytes.SIZEOF_LONG + Bytes.SIZEOF_INT;
+ }
+
+ private static byte[] encodeDecimal(Object object) {
+ if (object == null) {
+ return new byte[0];
+ }
+ BigDecimal v = (BigDecimal) object;
+ v = v.round(DEFAULT_MATH_CONTEXT).stripTrailingZeros();
+ int len = getLength(v);
+ byte[] result = new byte[Math.min(len, 21)];
+ decimalToBytes(v, result, 0, len);
+ return result;
+ }
+
+ private static BigDecimal decodeDecimal(byte[] bytes, int offset, int length) {
+ if (length == 1 && bytes[offset] == ZERO_BYTE) {
+ return BigDecimal.ZERO;
+ }
+ int signum = ((bytes[offset] & 0x80) == 0) ? -1 : 1;
+ int scale;
+ int index;
+ int digitOffset;
+ long multiplier = 100L;
+ int begIndex = offset + 1;
+ if (signum == 1) {
+ scale = (byte) (((bytes[offset] & 0x7F) - 65) * -2);
+ index = offset + length;
+ digitOffset = POS_DIGIT_OFFSET;
+ } else {
+ scale = (byte) ((~bytes[offset] - 65 - 128) * -2);
+ index = offset + length - (bytes[offset + length - 1] == NEG_TERMINAL_BYTE ? 1 : 0);
+ digitOffset = -NEG_DIGIT_OFFSET;
+ }
+ length = index - offset;
+ long l = signum * bytes[--index] - digitOffset;
+ if (l % 10 == 0) { // trailing zero
+ scale--; // drop trailing zero and compensate in the scale
+ l /= 10;
+ multiplier = 10;
+ }
+ // Use long arithmetic for as long as we can
+ while (index > begIndex) {
+ if (l >= MAX_LONG_FOR_DESERIALIZE || multiplier >= Long.MAX_VALUE / 100) {
+ multiplier = LongMath.divide(multiplier, 100L, RoundingMode.UNNECESSARY);
+ break; // Exit loop early so we don't overflow our multiplier
+ }
+ int digit100 = signum * bytes[--index] - digitOffset;
+ l += digit100 * multiplier;
+ multiplier = LongMath.checkedMultiply(multiplier, 100);
+ }
+
+ BigInteger bi;
+ // If still more digits, switch to BigInteger arithmetic
+ if (index > begIndex) {
+ bi = BigInteger.valueOf(l);
+ BigInteger biMultiplier = BigInteger.valueOf(multiplier).multiply(ONE_HUNDRED);
+ do {
+ int digit100 = signum * bytes[--index] - digitOffset;
+ bi = bi.add(biMultiplier.multiply(BigInteger.valueOf(digit100)));
+ biMultiplier = biMultiplier.multiply(ONE_HUNDRED);
+ } while (index > begIndex);
+ if (signum == -1) {
+ bi = bi.negate();
+ }
+ } else {
+ bi = BigInteger.valueOf(l * signum);
+ }
+ // Update the scale based on the precision
+ scale += (length - 2) * 2;
+ BigDecimal v = new BigDecimal(bi, scale);
+ return v;
+ }
+
+ private static int getLength(BigDecimal v) {
+ int signum = v.signum();
+ if (signum == 0) { // Special case for zero
+ return 1;
+ }
+ return (signum < 0 ? 2 : 1) + (v.precision() + 1 + (v.scale() % 2 == 0 ? 0 : 1)) / 2;
+ }
+
+ private static final int MAX_PRECISION = 38;
+ private static final MathContext DEFAULT_MATH_CONTEXT = new MathContext(MAX_PRECISION, RoundingMode.HALF_UP);
+ private static final Integer MAX_BIG_DECIMAL_BYTES = 21;
+ private static final byte ZERO_BYTE = (byte) 0x80;
+ private static final byte NEG_TERMINAL_BYTE = (byte) 102;
+ private static final int EXP_BYTE_OFFSET = 65;
+ private static final int POS_DIGIT_OFFSET = 1;
+ private static final int NEG_DIGIT_OFFSET = 101;
+ private static final BigInteger MAX_LONG = BigInteger.valueOf(Long.MAX_VALUE);
+ private static final BigInteger MIN_LONG = BigInteger.valueOf(Long.MIN_VALUE);
+ private static final BigInteger ONE_HUNDRED = BigInteger.valueOf(100);
+ private static final long MAX_LONG_FOR_DESERIALIZE = Long.MAX_VALUE / 1000;
+
+ private static int decimalToBytes(BigDecimal v, byte[] result, final int offset, int length) {
+ int signum = v.signum();
+ if (signum == 0) {
+ result[offset] = ZERO_BYTE;
+ return 1;
+ }
+ int index = offset + length;
+ int scale = v.scale();
+ int expOffset = scale % 2 * (scale < 0 ? -1 : 1);
+ int multiplyBy;
+ BigInteger divideBy;
+ if (expOffset == 0) {
+ multiplyBy = 1;
+ divideBy = ONE_HUNDRED;
+ } else {
+ multiplyBy = 10;
+ divideBy = BigInteger.TEN;
+ }
+ // Normalize the scale based on what is necessary to end up with a base 100
+ // decimal (i.e. 10.123e3)
+ int digitOffset;
+ BigInteger compareAgainst;
+ if (signum == 1) {
+ digitOffset = POS_DIGIT_OFFSET;
+ compareAgainst = MAX_LONG;
+ scale -= (length - 2) * 2;
+ result[offset] = (byte) ((-(scale + expOffset) / 2 + EXP_BYTE_OFFSET) | 0x80);
+ } else {
+ digitOffset = NEG_DIGIT_OFFSET;
+ compareAgainst = MIN_LONG;
+ // Scale adjustment shouldn't include terminal byte in length
+ scale -= (length - 2 - 1) * 2;
+ result[offset] = (byte) (~(-(scale + expOffset) / 2 + EXP_BYTE_OFFSET + 128) & 0x7F);
+ if (length <= MAX_BIG_DECIMAL_BYTES) {
+ result[--index] = NEG_TERMINAL_BYTE;
+ } else {
+ // Adjust length and offset down because we don't have enough room
+ length = MAX_BIG_DECIMAL_BYTES;
+ index = offset + length;
+ }
+ }
+ BigInteger bi = v.unscaledValue();
+ // Use BigDecimal arithmetic until we can fit into a long
+ while (bi.compareTo(compareAgainst) * signum > 0) {
+ BigInteger[] dandr = bi.divideAndRemainder(divideBy);
+ bi = dandr[0];
+ int digit = dandr[1].intValue();
+ result[--index] = (byte) (digit * multiplyBy + digitOffset);
+ multiplyBy = 1;
+ divideBy = ONE_HUNDRED;
+ }
+ long l = bi.longValue();
+ do {
+ long divBy = 100 / multiplyBy;
+ long digit = l % divBy;
+ l /= divBy;
+ result[--index] = (byte) (digit * multiplyBy + digitOffset);
+ multiplyBy = 1;
+ } while (l != 0);
+
+ return length;
+ }
+
+ private static void checkForSufficientLength(byte[] b, int offset, int requiredLength) {
+ if (b.length < offset + requiredLength) {
+ throw new RuntimeException(
+ "Expected length of at least " + requiredLength + " bytes, but had " + (b.length - offset));
+ }
+ }
+
+ private static Date parseDatetime(String dateStr) {
+ Date date = null;
+ int len = dateStr.length();
+ if (len == 10 && dateStr.charAt(4) == '-' && dateStr.charAt(7) == '-') {
+ date = new DateTime(dateStr).toDate();
+ } else if (len == 8 && dateStr.charAt(2) == ':' && dateStr.charAt(5) == ':') {
+ date = new DateTime("T" + dateStr).toDate();
+ } else if (len >= 19 && dateStr.charAt(4) == '-' && dateStr.charAt(7) == '-' && dateStr.charAt(13) == ':'
+ && dateStr.charAt(16) == ':') {
+ date = new DateTime(dateStr.replace(" ", "T")).toDate();
+ }
+ return date;
+ }
+}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/Type.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/Type.java
new file mode 100644
index 00000000..9a567191
--- /dev/null
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/Type.java
@@ -0,0 +1,104 @@
+package com.alibaba.otter.canal.client.adapter.hbase.support;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Date;
+
+/**
+ * Java类型
+ *
+ * @author machengyuan 2018-8-21 下午06:11:36
+ * @version 1.0.0
+ */
+public enum Type {
+ DEFAULT("STRING"),
+ STRING("STRING"),
+ INTEGER("INTEGER"),
+ LONG("LONG"),
+ SHORT("SHORT"),
+ BOOLEAN("BOOLEAN"),
+ FLOAT("FLOAT"),
+ DOUBLE("DOUBLE"),
+ BIGDECIMAL("BIGDECIMAL"),
+ DATE("DATE"),
+ BYTE("BYTE"),
+ BYTES("BYTES");
+
+ private String type;
+
+ Type(String type) {
+ this.type = type;
+ }
+
+ public String getType() {
+ return type;
+ }
+
+ public static Type getType(String type) {
+ if (type == null) {
+ return DEFAULT;
+ }
+ Type res;
+ if (type.equalsIgnoreCase("STRING")) {
+ res = STRING;
+ } else if (type.equalsIgnoreCase("INTEGER")) {
+ res = INTEGER;
+ } else if (type.equalsIgnoreCase("LONG")) {
+ res = LONG;
+ } else if (type.equalsIgnoreCase("SHORT")) {
+ res = SHORT;
+ } else if (type.equalsIgnoreCase("BOOLEAN")) {
+ res = BOOLEAN;
+ } else if (type.equalsIgnoreCase("FLOAT")) {
+ res = FLOAT;
+ } else if (type.equalsIgnoreCase("DOUBLE")) {
+ res = DOUBLE;
+ } else if (type.equalsIgnoreCase("BIGDECIMAL")) {
+ res = BIGDECIMAL;
+ } else if (type.equalsIgnoreCase("DATE")) {
+ res = DATE;
+ } else if (type.equalsIgnoreCase("BYTE")) {
+ res = BYTE;
+ } else if (type.equalsIgnoreCase("BYTES")) {
+ res = BYTES;
+ } else {
+ res = DEFAULT;
+ }
+ return res;
+ }
+
+ public static Type getType(Class javaType) {
+ if (javaType == null) {
+ return DEFAULT;
+ }
+ Type type;
+ if (Integer.class.isAssignableFrom(javaType) || int.class.isAssignableFrom(javaType)) {
+ type = INTEGER;
+ } else if (Long.class.isAssignableFrom(javaType) || long.class.isAssignableFrom(javaType)) {
+ type = LONG;
+ } else if (Byte.class.isAssignableFrom(javaType) || byte.class.isAssignableFrom(javaType)) {
+ type = BYTE;
+ } else if (Short.class.isAssignableFrom(javaType) || short.class.isAssignableFrom(javaType)) {
+ type = SHORT;
+ } else if (Float.class.isAssignableFrom(javaType) || float.class.isAssignableFrom(javaType)) {
+ type = FLOAT;
+ } else if (Double.class.isAssignableFrom(javaType) || double.class.isAssignableFrom(javaType)) {
+ type = DOUBLE;
+ } else if (Boolean.class.isAssignableFrom(javaType) || boolean.class.isAssignableFrom(javaType)) {
+ type = BOOLEAN;
+ } else if (Date.class.isAssignableFrom(javaType)) {
+ type = DATE;
+ } else if (byte[].class.isAssignableFrom(javaType)) {
+ type = BYTES;
+ } else if (String.class.isAssignableFrom(javaType)) {
+ type = STRING;
+ } else if (BigDecimal.class.isAssignableFrom(javaType)) {
+ type = BIGDECIMAL;
+ } else if (BigInteger.class.isAssignableFrom(javaType)) {
+ type = LONG;
+ } else {
+ type = DEFAULT;
+ }
+ return type;
+ }
+}
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java
new file mode 100644
index 00000000..07aad00f
--- /dev/null
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/TypeUtil.java
@@ -0,0 +1,188 @@
+package com.alibaba.otter.canal.client.adapter.hbase.support;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.Date;
+
+import org.apache.hadoop.hbase.util.Bytes;
+
+/**
+ * Java类型转换工具类
+ *
+ * @author machengyuan 2018-8-21 下午06:12:34
+ * @version 1.0.0
+ */
+public class TypeUtil {
+
+ public static byte[] toBytes(Object obj) {
+ if (obj == null) {
+ return null;
+ }
+ byte[] bytes;
+ if (obj instanceof String) {
+ bytes = Bytes.toBytes((String) obj);
+ } else if (obj instanceof Integer) {
+ bytes = Bytes.toBytes((Integer) obj);
+ } else if (obj instanceof Long) {
+ bytes = Bytes.toBytes((Long) obj);
+ } else if (obj instanceof Short) {
+ bytes = Bytes.toBytes((Short) obj);
+ } else if (obj instanceof Boolean) {
+ bytes = Bytes.toBytes((Boolean) obj);
+ } else if (obj instanceof Float) {
+ bytes = Bytes.toBytes((Float) obj);
+ } else if (obj instanceof Double) {
+ bytes = Bytes.toBytes((Double) obj);
+ } else if (obj instanceof BigDecimal) {
+ bytes = Bytes.toBytes((BigDecimal) obj);
+ } else if (obj instanceof BigInteger) {
+ bytes = Bytes.toBytes(((BigInteger) obj).longValue());
+ } else if (obj instanceof Date) {
+ bytes = Bytes.toBytes(((Date) obj).getTime());
+ } else if (obj instanceof Byte) {
+ bytes = new byte[] { (byte) obj };
+ } else if (obj instanceof byte[]) {
+ bytes = (byte[]) obj;
+ } else {
+ // 其余类型统一转换为string
+ bytes = Bytes.toBytes(obj.toString());
+ }
+ return bytes;
+ }
+
+ public static byte[] toBytes(Object v, Type type) {
+ if (v == null) {
+ return null;
+ }
+ byte[] b = null;
+ if (type == Type.DEFAULT) {
+ Type type1 = Type.getType(v.getClass());
+ if (type1 != null && type1 != Type.DEFAULT) {
+ b = toBytes(v, type1);
+ }
+ } else if (type == Type.STRING) {
+ b = Bytes.toBytes(v.toString());
+ } else if (type == Type.INTEGER) {
+ b = Bytes.toBytes(((Number) v).intValue());
+ } else if (type == Type.LONG) {
+ b = Bytes.toBytes(((Number) v).longValue());
+ } else if (type == Type.SHORT) {
+ b = Bytes.toBytes(((Number) v).shortValue());
+ } else if (type == Type.BYTE) {
+ b = Bytes.toBytes(((Number) v).byteValue());
+ } else if (type == Type.FLOAT) {
+ b = Bytes.toBytes(((Number) v).floatValue());
+ } else if (type == Type.DOUBLE) {
+ b = Bytes.toBytes(((Number) v).doubleValue());
+ } else if (type == Type.BOOLEAN) {
+ b = Bytes.toBytes(((Boolean) v));
+ } else if (type == Type.DATE) {
+ b = Bytes.toBytes(((Date) v).getTime());
+ } else if (type == Type.BYTES) {
+ b = (byte[]) v;
+ } else if (type == Type.BIGDECIMAL) {
+ if (v instanceof BigDecimal) {
+ b = Bytes.toBytes((BigDecimal) v);
+ } else {
+ b = Bytes.toBytes(new BigDecimal(v.toString()));
+ }
+ }
+ return b;
+ }
+
+ public static T toObject(byte[] bytes, Class clazz) {
+ if (bytes == null) {
+ return null;
+ }
+ Object res;
+ if (String.class.isAssignableFrom(clazz)) {
+ res = Bytes.toString(bytes);
+ } else if (Integer.class.isAssignableFrom(clazz)) {
+ res = Bytes.toInt(bytes);
+ } else if (Long.class.isAssignableFrom(clazz)) {
+ res = Bytes.toLong(bytes);
+ } else if (Short.class.isAssignableFrom(clazz)) {
+ res = Bytes.toShort(bytes);
+ } else if (Boolean.class.isAssignableFrom(clazz)) {
+ res = Bytes.toBoolean(bytes);
+ } else if (Float.class.isAssignableFrom(clazz)) {
+ res = Bytes.toFloat(bytes);
+ } else if (Double.class.isAssignableFrom(clazz)) {
+ res = Bytes.toDouble(bytes);
+ } else if (BigDecimal.class.isAssignableFrom(clazz)) {
+ res = Bytes.toBigDecimal(bytes);
+ } else if (BigInteger.class.isAssignableFrom(clazz)) {
+ res = Bytes.toLong(bytes);
+ } else if (java.sql.Date.class.isAssignableFrom(clazz)) {
+ long ts = Bytes.toLong(bytes);
+ res = new java.sql.Date(ts);
+ } else if (Time.class.isAssignableFrom(clazz)) {
+ long ts = Bytes.toLong(bytes);
+ res = new Time(ts);
+ } else if (Timestamp.class.isAssignableFrom(clazz)) {
+ long ts = Bytes.toLong(bytes);
+ res = new Timestamp(ts);
+ } else if (Date.class.isAssignableFrom(clazz)) {
+ long ts = Bytes.toLong(bytes);
+ res = new Date(ts);
+ } else if (Byte.class.isAssignableFrom(clazz)) {
+ res = bytes[0];
+ } else {
+ throw new IllegalArgumentException("mismatch class type");
+ }
+ // noinspection unchecked
+ return (T) res;
+ }
+
+ public static T toObject(byte[] bytes, Type type) {
+ if (bytes == null) {
+ return null;
+ }
+ Object res = null;
+ if (type == Type.STRING || type == Type.DEFAULT) {
+ res = Bytes.toString(bytes);
+ } else if (type == Type.INTEGER) {
+ if (bytes.length == Bytes.SIZEOF_INT) {
+ res = Bytes.toInt(bytes);
+ }
+ } else if (type == Type.LONG) {
+ if (bytes.length == Bytes.SIZEOF_LONG) {
+ res = Bytes.toLong(bytes);
+ }
+ } else if (type == Type.SHORT) {
+ if (bytes.length == Bytes.SIZEOF_SHORT) {
+ res = Bytes.toShort(bytes);
+ }
+ } else if (type == Type.BYTE) {
+ if (bytes.length == Bytes.SIZEOF_BYTE) {
+ res = bytes[0];
+ }
+ } else if (type == Type.FLOAT) {
+ if (bytes.length == Bytes.SIZEOF_FLOAT) {
+ res = Bytes.toFloat(bytes);
+ }
+ } else if (type == Type.DOUBLE) {
+ if (bytes.length == Bytes.SIZEOF_DOUBLE) {
+ res = Bytes.toDouble(bytes);
+ }
+ } else if (type == Type.BOOLEAN) {
+ if (bytes.length == Bytes.SIZEOF_BOOLEAN) {
+ res = Bytes.toBoolean(bytes);
+ }
+ } else if (type == Type.DATE) {
+ if (bytes.length == Bytes.SIZEOF_LONG) {
+ res = new Date(Bytes.toLong(bytes));
+ }
+ } else if (type == Type.BYTES) {
+ res = bytes;
+ } else if (type == Type.BIGDECIMAL) {
+ res = Bytes.toBigDecimal(bytes);
+ } else {
+ throw new IllegalArgumentException("mismatch class type");
+ }
+ // noinspection unchecked
+ return (T) res;
+ }
+}
diff --git a/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter b/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter
new file mode 100644
index 00000000..321d49cf
--- /dev/null
+++ b/client-adapter/hbase/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter
@@ -0,0 +1 @@
+hbase=com.alibaba.otter.canal.client.adapter.hbase.HbaseAdapter
\ No newline at end of file
diff --git a/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf b/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf
new file mode 100644
index 00000000..ff5b292e
--- /dev/null
+++ b/client-adapter/hbase/src/main/resources/hbase-mapping/configs.conf
@@ -0,0 +1,5 @@
+# 详细映射配置
+mytest_person2.yml
+
+# 简易配置, 只用指定数据库名.表名, 详细配置全部使用默认
+mytest.person
\ No newline at end of file
diff --git a/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml b/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml
new file mode 100644
index 00000000..575a1bf5
--- /dev/null
+++ b/client-adapter/hbase/src/main/resources/hbase-mapping/mytest_person2.yml
@@ -0,0 +1,56 @@
+hbaseOrm:
+ mode: PHOENIX #NATIVE #STRING
+ database: mytest # 数据库名
+ table: person2 # 数据库表名
+ hbaseTable: MYTEST.PERSON2 # HBase表名
+ family: CF # 默认统一Family名称
+ uppercaseQualifier: true # 字段名转大写, 默认为true
+ commitBatch: 3000 # 批量提交的大小
+ #rowKey: id,type # 复合字段rowKey不能和columns中的rowKey重复
+ columns:
+ # 数据库字段:HBase对应字段
+ id: ROWKEY$UNSIGNED_LONG
+ name: NAME
+ email: EMAIL
+ type: $DECIMAL
+ c_time: C_TIME$UNSIGNED_TIMESTAMP
+ birthday: BIRTHDAY$DATE
+
+# -- NATIVE类型
+# $DEFAULT
+# $STRING
+# $INTEGER
+# $LONG
+# $SHORT
+# $BOOLEAN
+# $FLOAT
+# $DOUBLE
+# $BIGDECIMAL
+# $DATE
+# $BYTE
+# $BYTES
+
+# -- PHOENIX类型
+# $DEFAULT 对应PHOENIX里的VARCHAR
+# $UNSIGNED_INT 对应PHOENIX里的UNSIGNED_INT 4字节
+# $UNSIGNED_LONG 对应PHOENIX里的UNSIGNED_LONG 8字节
+# $UNSIGNED_TINYINT 对应PHOENIX里的UNSIGNED_TINYINT 1字节
+# $UNSIGNED_SMALLINT 对应PHOENIX里的UNSIGNED_SMALLINT 2字节
+# $UNSIGNED_FLOAT 对应PHOENIX里的UNSIGNED_FLOAT 4字节
+# $UNSIGNED_DOUBLE 对应PHOENIX里的UNSIGNED_DOUBLE 8字节
+# $INTEGER 对应PHOENIX里的INTEGER 4字节
+# $BIGINT 对应PHOENIX里的BIGINT 8字节
+# $TINYINT 对应PHOENIX里的TINYINT 1字节
+# $SMALLINT 对应PHOENIX里的SMALLINT 2字节
+# $FLOAT 对应PHOENIX里的FLOAT 4字节
+# DOUBLE 对应PHOENIX里的DOUBLE 8字节
+# $BOOLEAN 对应PHOENIX里的BOOLEAN 1字节
+# $TIME 对应PHOENIX里的TIME 8字节
+# $DATE 对应PHOENIX里的DATE 8字节
+# $TIMESTAMP 对应PHOENIX里的TIMESTAMP 12字节
+# $UNSIGNED_TIME 对应PHOENIX里的UNSIGNED_TIME 8字节
+# $UNSIGNED_DATE 对应PHOENIX里的UNSIGNED_DATE 8字节
+# $UNSIGNED_TIMESTAMP 对应PHOENIX里的UNSIGNED_TIMESTAMP 12字节
+# $VARCHAR 对应PHOENIX里的VARCHAR 动态长度
+# $VARBINARY 对应PHOENIX里的VARBINARY 动态长度
+# $DECIMAL 对应PHOENIX里的DECIMAL 动态长度
\ No newline at end of file
diff --git a/client-adapter/logger/pom.xml b/client-adapter/logger/pom.xml
new file mode 100644
index 00000000..e6c5d218
--- /dev/null
+++ b/client-adapter/logger/pom.xml
@@ -0,0 +1,47 @@
+
+
+
+ canal.client-adapter
+ com.alibaba.otter
+ 1.1.0-SNAPSHOT
+
+ 4.0.0
+ com.alibaba.otter
+ client-adapter.logger
+ jar
+ canal client adapter logger example module for otter ${project.version}
+
+
+ com.alibaba.otter
+ client-adapter.common
+ ${project.version}
+ provided
+
+
+
+
+
+
+ org.apache.maven.plugins
+ maven-assembly-plugin
+ 2.4
+
+
+ jar-with-dependencies
+
+
+
+
+ make-assembly
+ package
+
+ single
+
+
+
+
+
+
+
\ No newline at end of file
diff --git a/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java
new file mode 100644
index 00000000..c6c1cb44
--- /dev/null
+++ b/client-adapter/logger/src/main/java/com/alibaba/otter/canal/client/adapter/logger/LoggerAdapterExample.java
@@ -0,0 +1,45 @@
+package com.alibaba.otter.canal.client.adapter.logger;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
+import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
+import com.alibaba.otter.canal.client.adapter.support.Dml;
+import com.alibaba.otter.canal.client.adapter.support.MessageUtil;
+import com.alibaba.otter.canal.client.adapter.support.SPI;
+import com.alibaba.otter.canal.protocol.Message;
+
+/**
+ * 外部适配器示例
+ *
+ * @author machengyuan 2018-8-19 下午11:45:38
+ * @version 1.0.0
+ */
+@SPI("logger") // logger参数对应CanalOuterAdapterConfiguration配置中的name
+public class LoggerAdapterExample implements CanalOuterAdapter {
+
+ private Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ @Override
+ public void writeOut(Message message) {
+ // 直接输出日志信息
+ MessageUtil.parse4Dml(message, new MessageUtil.Consumer() {
+
+ @Override
+ public void accept(Dml dml) {
+ logger.info(dml.toString());
+ }
+ });
+ }
+
+ @Override
+ public void init(CanalOuterAdapterConfiguration configuration) {
+
+ }
+
+ @Override
+ public void destroy() {
+
+ }
+}
diff --git a/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter b/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter
new file mode 100644
index 00000000..bfeae48e
--- /dev/null
+++ b/client-adapter/logger/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter
@@ -0,0 +1 @@
+logger=com.alibaba.otter.canal.client.adapter.logger.LoggerAdapterExample
\ No newline at end of file
diff --git a/client-adapter/pom.xml b/client-adapter/pom.xml
new file mode 100644
index 00000000..5398a9c5
--- /dev/null
+++ b/client-adapter/pom.xml
@@ -0,0 +1,21 @@
+
+
+
+ canal
+ com.alibaba.otter
+ 1.1.0-SNAPSHOT
+
+ 4.0.0
+ com.alibaba.otter
+ canal.client-adapter
+ pom
+ canal client adapter module for otter ${project.version}
+
+ common
+ logger
+ hbase
+
+
+
\ No newline at end of file
diff --git a/client-launcher/pom.xml b/client-launcher/pom.xml
index 42d605a7..952aeb8a 100644
--- a/client-launcher/pom.xml
+++ b/client-launcher/pom.xml
@@ -13,6 +13,12 @@
jar
canal client launcher module for otter ${project.version}
+
+ com.alibaba.otter
+ client-adapter.common
+ ${project.version}
+
+
com.alibaba.otter
canal.client
@@ -24,16 +30,25 @@
canal.kafka.client
${project.version}
-
- joda-time
- joda-time
- 2.9.4
-
org.yaml
snakeyaml
1.17
+
+
+
+ com.alibaba.otter
+ client-adapter.logger
+ ${project.version}
+ jar-with-dependencies
+
+
+ com.alibaba.otter
+ client-adapter.hbase
+ ${project.version}
+ jar-with-dependencies
+
diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java
index 556910dd..c94fcd28 100644
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java
+++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/ClientLauncher.java
@@ -1,6 +1,7 @@
package com.alibaba.otter.canal.client;
-import com.alibaba.otter.canal.client.support.CanalClientConfig;
+import com.alibaba.otter.canal.client.adapter.loader.CanalAdapterLoader;
+import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -9,8 +10,9 @@ import org.yaml.snakeyaml.Yaml;
import java.io.FileInputStream;
public class ClientLauncher {
+
private static final String CLASSPATH_URL_PREFIX = "classpath:";
- private static final Logger logger = LoggerFactory.getLogger(ClientLauncher.class);
+ private static final Logger logger = LoggerFactory.getLogger(ClientLauncher.class);
public static void main(String[] args) {
try {
@@ -23,7 +25,7 @@ public class ClientLauncher {
if (conf.startsWith(CLASSPATH_URL_PREFIX)) {
conf = StringUtils.substringAfter(conf, CLASSPATH_URL_PREFIX);
canalClientConfig = new Yaml().loadAs(ClientLauncher.class.getClassLoader().getResourceAsStream(conf),
- CanalClientConfig.class);
+ CanalClientConfig.class);
} else {
canalClientConfig = new Yaml().loadAs(new FileInputStream(conf), CanalClientConfig.class);
}
diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/AbstractCanalAdapterWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/AbstractCanalAdapterWorker.java
deleted file mode 100644
index 42a154d4..00000000
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/AbstractCanalAdapterWorker.java
+++ /dev/null
@@ -1,156 +0,0 @@
-package com.alibaba.otter.canal.client.adapter;
-
-import com.alibaba.fastjson.JSON;
-import com.alibaba.otter.canal.client.support.Dml;
-import com.alibaba.otter.canal.client.support.JdbcTypeUtil;
-import com.alibaba.otter.canal.protocol.CanalEntry;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.util.*;
-import java.util.concurrent.Callable;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Future;
-
-/**
- * 适配器工作线程抽象类
- *
- * @author machengyuan 2018-8-19 下午11:30:49
- * @version 1.0.0
- */
-public abstract class AbstractCanalAdapterWorker {
- protected final Logger logger = LoggerFactory.getLogger(this.getClass());
-
- protected String canalDestination; // canal实例
- protected List> canalOuterAdapters; // 外部适配器
- protected ExecutorService groupInnerExecutorService; // 组内工作线程池
- protected volatile boolean running = false; // 是否运行中
- protected Thread thread = null;
- protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
- @Override
- public void uncaughtException(Thread t, Throwable e) {
- logger.error("parse events has an error", e);
- }
- };
-
- /**
- * 将每个entry转换为DML操作对象并调用所有适配器写入
- *
- * @param entries
- */
- protected void convertAndWrite(List entries) {
- for (CanalEntry.Entry entry : entries) {
- if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
- continue;
- }
-
- CanalEntry.RowChange rowChange;
- try {
- rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
- } catch (Exception e) {
- throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
- }
-
- CanalEntry.EventType eventType = rowChange.getEventType();
-
- final Dml dml = new Dml();
- dml.setCanalDestination(canalDestination);
- dml.setDatabase(entry.getHeader().getSchemaName());
- dml.setTable(entry.getHeader().getTableName());
- dml.setType(eventType.toString());
- dml.setTs(System.currentTimeMillis());
- dml.setSql(rowChange.getSql());
- List> data = new ArrayList<>();
- List> old = new ArrayList<>();
-
- if (!rowChange.getIsDdl()) {
- Set updateSet = new HashSet<>();
- for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
- Map row = new LinkedHashMap<>();
- List columns;
-
- if (eventType == CanalEntry.EventType.DELETE) {
- columns = rowData.getBeforeColumnsList();
- } else {
- columns = rowData.getAfterColumnsList();
- }
-
- for (CanalEntry.Column column : columns) {
- row.put(column.getName(), JdbcTypeUtil.typeConvert(dml.getTable(), column.getName(), column.getValue(),
- column.getSqlType(), column.getMysqlType()));
- //获取update为true的字段
- if (column.getUpdated()) {
- updateSet.add(column.getName());
- }
- }
- if (!row.isEmpty()) {
- data.add(row);
- }
-
- if (eventType == CanalEntry.EventType.UPDATE) {
- Map rowOld = new LinkedHashMap<>();
- for (CanalEntry.Column column : rowData.getBeforeColumnsList()) {
- if (updateSet.contains(column.getName())) {
- rowOld.put(column.getName(), JdbcTypeUtil.typeConvert(dml.getTable(), column.getName(),
- column.getValue(), column.getSqlType(), column.getMysqlType()));
- }
- }
- // update操作将记录修改前的值
- if (!rowOld.isEmpty()) {
- old.add(rowOld);
- }
- }
- }
- if (!data.isEmpty()) {
- dml.setData(data);
- }
- if (!old.isEmpty()) {
- dml.setOld(old);
- }
- }
- List> futures = new ArrayList<>();
- // 组间适配器并行运行
- for (List outerAdapters : canalOuterAdapters) {
- final List adapters = outerAdapters;
- futures.add(groupInnerExecutorService.submit(
- new Callable() {
- @Override
- public Boolean call() {
- boolean flag = true;
- // 组内适配器穿行运行，尽量不要配置组内适配器
- for (CanalOuterAdapter c : adapters) {
- long begin = System.currentTimeMillis();
- if (!c.writeOut(dml)) {
- logger.error("{} write fail！ data: {}", c.getClass().getName(), JSON.toJSONString(dml));
-
- flag = false;
- }
- if (logger.isDebugEnabled()) {
- logger.debug("{} elapsed time: {}", c.getClass().getName(), (System.currentTimeMillis() - begin));
- }
- }
- return flag;
- }
- })
- );
-
- // 等待所有适配器写入完成
- // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
- for (Future f : futures) {
- try {
- if (!f.get()) {
- logger.error("Outer adapter write failed");
- }
- } catch (InterruptedException | ExecutionException e) {
- logger.error(e.getMessage(), e);
- }
- }
- }
- }
- }
-
- public abstract void start();
-
- public abstract void stop();
-}
\ No newline at end of file
diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java
new file mode 100644
index 00000000..9dd328e3
--- /dev/null
+++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/AbstractCanalAdapterWorker.java
@@ -0,0 +1,83 @@
+package com.alibaba.otter.canal.client.adapter.loader;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+
+import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.alibaba.otter.canal.protocol.Message;
+
+/**
+ * 适配器工作线程抽象类
+ *
+ * @author machengyuan 2018-8-19 下午11:30:49
+ * @version 1.0.0
+ */
+public abstract class AbstractCanalAdapterWorker {
+
+ protected final Logger logger = LoggerFactory.getLogger(this.getClass());
+
+ protected String canalDestination; // canal实例
+ protected List> canalOuterAdapters; // 外部适配器
+ protected ExecutorService groupInnerExecutorService; // 组内工作线程池
+ protected volatile boolean running = false; // 是否运行中
+ protected Thread thread = null;
+ protected Thread.UncaughtExceptionHandler handler = new Thread.UncaughtExceptionHandler() {
+
+ @Override
+ public void uncaughtException(Thread t, Throwable e) {
+ logger.error("parse events has an error", e);
+ }
+ };
+
+ protected void writeOut(final Message message) {
+ List> futures = new ArrayList<>();
+ // 组间适配器并行运行
+ for (List outerAdapters : canalOuterAdapters) {
+ final List adapters = outerAdapters;
+ futures.add(groupInnerExecutorService.submit(new Callable() {
+
+ @Override
+ public Boolean call() {
+ try {
+ // 组内适配器穿行运行，尽量不要配置组内适配器
+ for (CanalOuterAdapter c : adapters) {
+ long begin = System.currentTimeMillis();
+ c.writeOut(message);
+ if (logger.isDebugEnabled()) {
+ logger.debug("{} elapsed time: {}",
+ c.getClass().getName(),
+ (System.currentTimeMillis() - begin));
+ }
+ }
+ return true;
+ } catch (Exception e) {
+ return false;
+ }
+ }
+ }));
+
+ // 等待所有适配器写入完成
+ // 由于是组间并发操作，所以将阻塞直到耗时最久的工作组操作完成
+ for (Future f : futures) {
+ try {
+ if (!f.get()) {
+ logger.error("Outer adapter write failed");
+ }
+ } catch (InterruptedException | ExecutionException e) {
+ logger.error(e.getMessage(), e);
+ }
+ }
+ }
+ }
+
+ public abstract void start();
+
+ public abstract void stop();
+}
diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/CanalAdapterKafkaWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java
similarity index 62%
rename from client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/CanalAdapterKafkaWorker.java
rename to client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java
index 2e64f8b8..6b35e452 100644
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/CanalAdapterKafkaWorker.java
+++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterKafkaWorker.java
@@ -1,5 +1,7 @@
-package com.alibaba.otter.canal.client.adapter;
+package com.alibaba.otter.canal.client.adapter.loader;
+import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
+import com.alibaba.otter.canal.client.adapter.loader.AbstractCanalAdapterWorker;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnector;
import com.alibaba.otter.canal.kafka.client.KafkaCanalConnectors;
import com.alibaba.otter.canal.protocol.Message;
@@ -13,11 +15,13 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
+
private KafkaCanalConnector connector;
- private String topic;
+ private String topic;
- public CanalAdapterKafkaWorker(String zkServers, String bootstrapServers, String topic, String groupId, List> canalOuterAdapters) {
+ public CanalAdapterKafkaWorker(String zkServers, String bootstrapServers, String topic, String groupId,
+ List> canalOuterAdapters){
this.canalOuterAdapters = canalOuterAdapters;
this.groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
this.topic = topic;
@@ -25,13 +29,14 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
connector = KafkaCanalConnectors.newKafkaConnector(zkServers, bootstrapServers, topic, null, groupId);
// connector.setSessionTimeout(5L, TimeUnit.MINUTES);
-// super.initSwitcher(topic);
+ // super.initSwitcher(topic);
}
@Override
public void start() {
if (!running) {
thread = new Thread(new Runnable() {
+
@Override
public void run() {
process();
@@ -53,9 +58,9 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
connector.stopRunning();
running = false;
-// if (switcher != null && !switcher.state()) {
-// switcher.set(true);
-// }
+ // if (switcher != null && !switcher.state()) {
+ // switcher.set(true);
+ // }
if (thread != null) {
try {
@@ -78,7 +83,8 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
}
private void process() {
- while (!running) ;
+ while (!running)
+ ;
ExecutorService executor = Executors.newFixedThreadPool(1);
final AtomicBoolean executing = new AtomicBoolean(true);
while (running) {
@@ -90,37 +96,45 @@ public class CanalAdapterKafkaWorker extends AbstractCanalAdapterWorker {
logger.info("=============> Subscribe topic: {} succeed<=============", this.topic);
while (running) {
try {
-// switcher.get(); //等待开关开启
+ // switcher.get(); //等待开关开启
final Message message = connector.getWithoutAck();
executing.set(true);
if (message != null) {
- executor.submit(
- new Runnable() {
- @Override
- public void run() {
- try {
- if (logger.isDebugEnabled()) {
- logger.debug("topic: {} batchId: {} batchSize: {} ", topic, message.getId(), message.getEntries().size());
- }
- long begin = System.currentTimeMillis();
- convertAndWrite(message.getEntries());
- long now = System.currentTimeMillis();
- if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) {
- logger.error("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
- }
- if (logger.isDebugEnabled()) {
- logger.debug("topic: {} batchId {} elapsed time: {} ms", topic, message.getId(), now - begin);
- }
- } catch (Exception e) {
- logger.error(e.getMessage(), e);
- } finally {
- executing.compareAndSet(true, false);
- }
+ executor.submit(new Runnable() {
+
+ @Override
+ public void run() {
+ try {
+ if (logger.isDebugEnabled()) {
+ logger.debug("topic: {} batchId: {} batchSize: {} ",
+ topic,
+ message.getId(),
+ message.getEntries().size());
}
+ long begin = System.currentTimeMillis();
+ writeOut(message);
+ long now = System.currentTimeMillis();
+ if ((System.currentTimeMillis() - begin) > 5 * 60 * 1000) {
+ logger.error("topic: {} batchId {} elapsed time: {} ms",
+ topic,
+ message.getId(),
+ now - begin);
+ }
+ if (logger.isDebugEnabled()) {
+ logger.debug("topic: {} batchId {} elapsed time: {} ms",
+ topic,
+ message.getId(),
+ now - begin);
+ }
+ } catch (Exception e) {
+ logger.error(e.getMessage(), e);
+ } finally {
+ executing.compareAndSet(true, false);
}
- );
+ }
+ });
while (executing.get()) { // keeping kafka client active
connector.ack();
diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/CanalAdapterLoader.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java
similarity index 82%
rename from client-launcher/src/main/java/com/alibaba/otter/canal/client/CanalAdapterLoader.java
rename to client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java
index f60a830e..b0fe35c5 100644
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/CanalAdapterLoader.java
+++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterLoader.java
@@ -1,11 +1,9 @@
-package com.alibaba.otter.canal.client;
+package com.alibaba.otter.canal.client.adapter.loader;
-import com.alibaba.otter.canal.client.adapter.CanalAdapterKafkaWorker;
-import com.alibaba.otter.canal.client.adapter.CanalAdapterWorker;
import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
-import com.alibaba.otter.canal.client.adapter.CanalOuterAdapterConfiguration;
-import com.alibaba.otter.canal.client.support.CanalClientConfig;
-import com.alibaba.otter.canal.client.support.ExtensionLoader;
+import com.alibaba.otter.canal.client.adapter.support.CanalClientConfig;
+import com.alibaba.otter.canal.client.adapter.support.CanalOuterAdapterConfiguration;
+import com.alibaba.otter.canal.client.adapter.support.ExtensionLoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -25,17 +23,18 @@ import java.util.concurrent.Executors;
* @version 1.0.0
*/
public class CanalAdapterLoader {
- private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class);
+
+ private static final Logger logger = LoggerFactory.getLogger(CanalAdapterLoader.class);
private CanalClientConfig canalClientConfig;
- private Map canalWorkers = new HashMap<>();
+ private Map canalWorkers = new HashMap<>();
private Map canalKafkaWorkers = new HashMap<>();
private ExtensionLoader loader;
- public CanalAdapterLoader(CanalClientConfig canalClientConfig) {
+ public CanalAdapterLoader(CanalClientConfig canalClientConfig){
this.canalClientConfig = canalClientConfig;
}
@@ -48,7 +47,8 @@ public class CanalAdapterLoader {
throw new RuntimeException("Blank config property: canalInstances or canalKafkaTopics");
}
- loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class, "" /*TODO canalClientConfig.getClassloaderPolicy()*/);
+ loader = ExtensionLoader.getExtensionLoader(CanalOuterAdapter.class,
+ "" /* TODO canalClientConfig.getClassloaderPolicy() */);
String canalServerHost = this.canalClientConfig.getCanalServerHost();
SocketAddress sa = null;
@@ -91,20 +91,23 @@ public class CanalAdapterLoader {
List canalOuterAdapters = new ArrayList<>();
- for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) {
-// for (CanalOuterAdapterConfiguration config : adaptor.getOutAdapters()) {
- loadConnector(config, canalOuterAdapters);
-// }
+ for (CanalOuterAdapterConfiguration config : group.getOutAdapters()) {
+ // for (CanalOuterAdapterConfiguration config : adaptor.getOutAdapters()) {
+ loadConnector(config, canalOuterAdapters);
+ // }
}
canalOuterAdapterGroups.add(canalOuterAdapters);
String zkServers = canalClientConfig.getZookeeperHosts();
CanalAdapterKafkaWorker canalKafkaWorker = new CanalAdapterKafkaWorker(zkServers,
- canalClientConfig.getBootstrapServers(),
- kafkaTopic.getTopic(), group.getGroupId(), canalOuterAdapterGroups);
+ canalClientConfig.getBootstrapServers(),
+ kafkaTopic.getTopic(),
+ group.getGroupId(),
+ canalOuterAdapterGroups);
canalKafkaWorkers.put(kafkaTopic.getTopic() + "-" + group.getGroupId(), canalKafkaWorker);
canalKafkaWorker.start();
- logger.info("Start adapter for canal-client kafka topic: {} succeed", kafkaTopic.getTopic() + "-" + group.getGroupId());
+ logger.info("Start adapter for canal-client kafka topic: {} succeed",
+ kafkaTopic.getTopic() + "-" + group.getGroupId());
}
}
}
@@ -125,8 +128,7 @@ public class CanalAdapterLoader {
}
/**
- * 销毁所有适配器
- * 为防止canal实例太多造成销毁阻塞, 并行销毁
+ * 销毁所有适配器 为防止canal实例太多造成销毁阻塞, 并行销毁
*/
public void destroy() {
if (canalWorkers.size() > 0) {
@@ -134,6 +136,7 @@ public class CanalAdapterLoader {
for (CanalAdapterWorker v : canalWorkers.values()) {
final CanalAdapterWorker caw = v;
stopExecutorService.submit(new Runnable() {
+
@Override
public void run() {
caw.stop();
@@ -147,6 +150,7 @@ public class CanalAdapterLoader {
for (CanalAdapterKafkaWorker v : canalKafkaWorkers.values()) {
final CanalAdapterKafkaWorker cakw = v;
stopKafkaExecutorService.submit(new Runnable() {
+
@Override
public void run() {
cakw.stop();
diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/CanalAdapterWorker.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java
similarity index 74%
rename from client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/CanalAdapterWorker.java
rename to client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java
index 2af88798..bca40041 100644
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/CanalAdapterWorker.java
+++ b/client-launcher/src/main/java/com/alibaba/otter/canal/client/adapter/loader/CanalAdapterWorker.java
@@ -1,7 +1,9 @@
-package com.alibaba.otter.canal.client.adapter;
+package com.alibaba.otter.canal.client.adapter.loader;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
+import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
+import com.alibaba.otter.canal.client.adapter.loader.AbstractCanalAdapterWorker;
import com.alibaba.otter.canal.client.impl.ClusterCanalConnector;
import com.alibaba.otter.canal.protocol.Message;
@@ -16,19 +18,21 @@ import java.util.concurrent.Executors;
* @version 1.0.0
*/
public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
+
private static final int BATCH_SIZE = 50;
private static final int SO_TIMEOUT = 0;
- private CanalConnector connector;
+ private CanalConnector connector;
/**
* 单台client适配器worker的构造方法
*
- * @param canalDestination canal实例名
- * @param address canal-server地址
+ * @param canalDestination canal实例名
+ * @param address canal-server地址
* @param canalOuterAdapters 外部适配器组
*/
- public CanalAdapterWorker(String canalDestination, SocketAddress address, List> canalOuterAdapters) {
+ public CanalAdapterWorker(String canalDestination, SocketAddress address,
+ List> canalOuterAdapters){
this.canalOuterAdapters = canalOuterAdapters;
this.canalDestination = canalDestination;
groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
@@ -38,24 +42,26 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
/**
* HA模式下client适配器worker的构造方法
*
- * @param canalDestination canal实例名
- * @param zookeeperHosts zookeeper地址
+ * @param canalDestination canal实例名
+ * @param zookeeperHosts zookeeper地址
* @param canalOuterAdapters 外部适配器组
*/
- public CanalAdapterWorker(String canalDestination, String zookeeperHosts, List> canalOuterAdapters) {
+ public CanalAdapterWorker(String canalDestination, String zookeeperHosts,
+ List> canalOuterAdapters){
this.canalOuterAdapters = canalOuterAdapters;
this.canalDestination = canalDestination;
groupInnerExecutorService = Executors.newFixedThreadPool(canalOuterAdapters.size());
connector = CanalConnectors.newClusterConnector(zookeeperHosts, canalDestination, "", "");
((ClusterCanalConnector) connector).setSoTimeout(SO_TIMEOUT);
-// super.initSwitcher(canalDestination);
+ // super.initSwitcher(canalDestination);
}
@Override
public void start() {
if (!running) {
thread = new Thread(new Runnable() {
+
@Override
public void run() {
process();
@@ -74,9 +80,9 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
return;
}
-// if (switcher != null && !switcher.state()) {
-// switcher.set(true);
-// }
+ // if (switcher != null && !switcher.state()) {
+ // switcher.set(true);
+ // }
connector.stopRunning();
running = false;
@@ -103,27 +109,29 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
}
private void process() {
- while (!running) ; //waiting until running == true
+ while (!running)
+ ; // waiting until running == true
while (running) {
try {
-// if (switcher != null) {
-// switcher.get();
-// }
+ // if (switcher != null) {
+ // switcher.get();
+ // }
logger.info("=============> Start to connect destination: {} <=============", this.canalDestination);
connector.connect();
logger.info("=============> Start to subscribe destination: {} <=============", this.canalDestination);
connector.subscribe();
logger.info("=============> Subscribe destination: {} succeed <=============", this.canalDestination);
while (running) {
-// try {
-// if (switcher != null) {
-// switcher.get();
-// }
-// } catch (TimeoutException e) {
-// break;
-// }
+ // try {
+ // if (switcher != null) {
+ // switcher.get();
+ // }
+ // } catch (TimeoutException e) {
+ // break;
+ // }
- // server配置canal.instance.network.soTimeout(默认: 30s) 范围内未与server交互，server将关闭本次socket连接
+ // server配置canal.instance.network.soTimeout(默认: 30s)
+ // 范围内未与server交互，server将关闭本次socket连接
Message message = connector.getWithoutAck(BATCH_SIZE); // 获取指定数量的数据
long batchId = message.getId();
try {
@@ -133,17 +141,23 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
- //ignore
+ // ignore
}
} else {
if (logger.isDebugEnabled()) {
- logger.debug("destination: {} batchId: {} batchSize: {} ", this.canalDestination, batchId, size);
+ logger.debug("destination: {} batchId: {} batchSize: {} ",
+ this.canalDestination,
+ batchId,
+ size);
}
long begin = System.currentTimeMillis();
- convertAndWrite(message.getEntries());
+ writeOut(message);
long now = System.currentTimeMillis();
if (logger.isDebugEnabled()) {
- logger.debug("destination: {} batchId: {} elapsed time: {} ms", this.canalDestination, batchId, now - begin);
+ logger.debug("destination: {} batchId: {} elapsed time: {} ms",
+ this.canalDestination,
+ batchId,
+ now - begin);
}
}
connector.ack(batchId); // 提交确认
@@ -153,7 +167,6 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
}
}
-
} catch (Exception e) {
logger.error("process error!", e);
} finally {
@@ -161,11 +174,11 @@ public class CanalAdapterWorker extends AbstractCanalAdapterWorker {
logger.info("=============> Disconnect destination: {} <=============", this.canalDestination);
}
- if (running) { //is reconnect
+ if (running) { // is reconnect
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
- //ignore
+ // ignore
}
}
diff --git a/client-launcher/src/main/java/com/alibaba/otter/canal/client/example/LoggerAdapterExample.java b/client-launcher/src/main/java/com/alibaba/otter/canal/client/example/LoggerAdapterExample.java
deleted file mode 100644
index 0f2b4612..00000000
--- a/client-launcher/src/main/java/com/alibaba/otter/canal/client/example/LoggerAdapterExample.java
+++ /dev/null
@@ -1,36 +0,0 @@
-package com.alibaba.otter.canal.client.example;
-
-import com.alibaba.otter.canal.client.adapter.CanalOuterAdapter;
-import com.alibaba.otter.canal.client.adapter.CanalOuterAdapterConfiguration;
-import com.alibaba.otter.canal.client.support.Dml;
-import com.alibaba.otter.canal.client.support.SPI;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * 外部适配器示例
- *
- * @author machengyuan 2018-8-19 下午11:45:38
- * @version 1.0.0
- */
-@SPI("logger") // logger参数对应CanalOuterAdapterConfiguration配置中的name
-public class LoggerAdapterExample implements CanalOuterAdapter {
- private Logger logger = LoggerFactory.getLogger(this.getClass());
-
- @Override
- public Boolean writeOut(Dml dml) {
- // 直接输出一个日志信息
- logger.info(dml.toString());
- return true;
- }
-
- @Override
- public void init(CanalOuterAdapterConfiguration configuration) {
-
- }
-
- @Override
- public void destroy() {
-
- }
-}
diff --git a/client-launcher/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter b/client-launcher/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter
deleted file mode 100644
index 787f15d7..00000000
--- a/client-launcher/src/main/resources/META-INF/canal/com.alibaba.otter.canal.client.adapter.CanalOuterAdapter
+++ /dev/null
@@ -1 +0,0 @@
-logger=com.alibaba.otter.canal.client.example.LoggerAdapterExample
\ No newline at end of file
diff --git a/client-launcher/src/main/resources/canal-client.yml b/client-launcher/src/main/resources/canal-client.yml
index 80a912a2..2412e38d 100644
--- a/client-launcher/src/main/resources/canal-client.yml
+++ b/client-launcher/src/main/resources/canal-client.yml
@@ -1,5 +1,5 @@
-#canalServerHost: 127.0.0.1:11111
-zookeeperHosts: 127.0.0.1:2181
+canalServerHost: 127.0.0.1:11111
+#zookeeperHosts: 127.0.0.1:2181
#bootstrapServers: kafka1.mytest.com:9092,kafka2.mytest.com:9092
canalInstances:
@@ -7,9 +7,9 @@ canalInstances:
adapterGroups:
- outAdapters:
- name: logger
-# - name: hbase
-# hosts:
-# properties: {znodeParent: /hbase}
+ - name: hbase
+ hosts: slave1:2181
+ properties: {znodeParent: "/hbase-unsecure"}
#kafkaTopics:
#- topic: devmysql4308
diff --git a/client-launcher/src/main/resources/logback.xml b/client-launcher/src/main/resources/logback.xml
index 41979ba0..665954f7 100644
--- a/client-launcher/src/main/resources/logback.xml
+++ b/client-launcher/src/main/resources/logback.xml
@@ -36,12 +36,12 @@
-
+
-
+
\ No newline at end of file
diff --git a/pom.xml b/pom.xml
index 19b07a06..e7bb158e 100644
--- a/pom.xml
+++ b/pom.xml
@@ -120,6 +120,7 @@
kafka
kafka-client
prometheus
+ client-adapter
client-launcher
@@ -333,7 +334,7 @@
org.apache.maven.plugins
maven-compiler-plugin
- 3.8.0
+ 3.7.0
${java_source_version}
${java_target_version}