chenjl/canal
1
0
Fork 0
mirror of https://github.com/alibaba/canal.git synced 2026-11-04 00:30:15 +08:00
Code Issues Packages Projects Releases Wiki Activity

修复adapter无法回退的问题，fixed 1341

Browse Source 
修复HBase断连的问题
This commit is contained in:
mcy
2018-12-26 15:24:30 +08:00
parent dffbc437a6
commit 9e6adff942
7 changed files with 197 additions and 202 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/service/ESSyncService.java
View File

@@ -102,7 +102,7 @@ public class ESSyncService {
}
} catch (Exception e) {
logger.error("sync error, es index: {}, DML : {}", config.getEsMapping().get_index(), dml);
logger.error(e.getMessage(), e);
throw new RuntimeException(e);
}
}

18
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
View File

@@ -13,8 +13,6 @@ import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
@@ -30,11 +28,7 @@ import com.alibaba.otter.canal.client.adapter.hbase.monitor.HbaseConfigMonitor;
import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseEtlService;
import com.alibaba.otter.canal.client.adapter.hbase.service.HbaseSyncService;
import com.alibaba.otter.canal.client.adapter.hbase.support.HbaseTemplate;
import com.alibaba.otter.canal.client.adapter.support.DatasourceConfig;
import com.alibaba.otter.canal.client.adapter.support.Dml;
import com.alibaba.otter.canal.client.adapter.support.EtlResult;
import com.alibaba.otter.canal.client.adapter.support.OuterAdapterConfig;
import com.alibaba.otter.canal.client.adapter.support.SPI;
import com.alibaba.otter.canal.client.adapter.support.*;
/**
* HBase外部适配器
@@ -50,7 +44,6 @@ public class HbaseAdapter implements OuterAdapter {
private Map<String, MappingConfig> hbaseMapping = new ConcurrentHashMap<>(); // 文件名对应配置
private Map<String, Map<String, MappingConfig>> mappingConfigCache = new ConcurrentHashMap<>(); // 库名-表名对应配置
private Connection conn;
private HbaseSyncService hbaseSyncService;
private HbaseTemplate hbaseTemplate;
@@ -90,8 +83,7 @@ public class HbaseAdapter implements OuterAdapter {
Configuration hbaseConfig = HBaseConfiguration.create();
properties.forEach(hbaseConfig::set);
conn = ConnectionFactory.createConnection(hbaseConfig);
hbaseTemplate = new HbaseTemplate(conn);
hbaseTemplate = new HbaseTemplate(hbaseConfig);
hbaseSyncService = new HbaseSyncService(hbaseTemplate);
configMonitor = new HbaseConfigMonitor();
@@ -174,7 +166,7 @@ public class HbaseAdapter implements OuterAdapter {
String hbaseTable = config.getHbaseMapping().getHbaseTable();
long rowCount = 0L;
try {
HTable table = (HTable) conn.getTable(TableName.valueOf(hbaseTable));
HTable table = (HTable) hbaseTemplate.getConnection().getTable(TableName.valueOf(hbaseTable));
Scan scan = new Scan();
scan.setFilter(new FirstKeyOnlyFilter());
ResultScanner resultScanner = table.getScanner(scan);
@@ -195,14 +187,12 @@ public class HbaseAdapter implements OuterAdapter {
if (configMonitor != null) {
configMonitor.destroy();
}
if (conn != null) {
try {
conn.close();
hbaseTemplate.close();
} catch (IOException e) {
throw new RuntimeException(e);
}
}
}
@Override
public String getDestination(String task) {

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

@@ -29,7 +29,6 @@ public class HbaseSyncService {
}
public void sync(MappingConfig config, Dml dml) {
try {
if (config != null) {
String type = dml.getType();
if (type != null && type.equalsIgnoreCase("INSERT")) {
@@ -43,9 +42,6 @@ public class HbaseSyncService {
logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
/**

50
client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/support/HbaseTemplate.java
View File

@@ -5,14 +5,11 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -27,14 +24,31 @@ public class HbaseTemplate {
private Logger logger = LoggerFactory.getLogger(this.getClass());
private Connection conn;
private Configuration hbaseConfig; // hbase配置对象
private Connection conn; // hbase连接
public HbaseTemplate(Connection conn){
this.conn = conn;
public HbaseTemplate(Configuration hbaseConfig){
this.hbaseConfig = hbaseConfig;
initConn();
}
private void initConn() {
try {
this.conn = ConnectionFactory.createConnection(hbaseConfig);
} catch (IOException e) {
throw new RuntimeException(e);
}
}
public Connection getConnection() {
if (conn == null || conn.isAborted() || conn.isClosed()) {
initConn();
}
return conn;
}
public boolean tableExists(String tableName) {
try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
try (HBaseAdmin admin = (HBaseAdmin) getConnection().getAdmin()) {
return admin.tableExists(TableName.valueOf(tableName));
} catch (IOException e) {
@@ -43,7 +57,7 @@ public class HbaseTemplate {
}
public void createTable(String tableName, String... familyNames) {
try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
try (HBaseAdmin admin = (HBaseAdmin) getConnection().getAdmin()) {
HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
// 添加列簇
@@ -60,7 +74,7 @@ public class HbaseTemplate {
}
public void disableTable(String tableName) {
try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
try (HBaseAdmin admin = (HBaseAdmin) getConnection().getAdmin()) {
admin.disableTable(tableName);
} catch (IOException e) {
logger.error(e.getMessage(), e);
@@ -69,7 +83,7 @@ public class HbaseTemplate {
}
public void deleteTable(String tableName) {
try (HBaseAdmin admin = (HBaseAdmin) conn.getAdmin()) {
try (HBaseAdmin admin = (HBaseAdmin) getConnection().getAdmin()) {
if (admin.isTableEnabled(tableName)) {
disableTable(tableName);
}
@@ -90,7 +104,7 @@ public class HbaseTemplate {
public Boolean put(String tableName, HRow hRow) {
boolean flag = false;
try {
HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
HTable table = (HTable) getConnection().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());
@@ -114,7 +128,7 @@ public class HbaseTemplate {
public Boolean puts(String tableName, List<HRow> rows) {
boolean flag = false;
try {
HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
HTable table = (HTable) getConnection().getTable(TableName.valueOf(tableName));
List<Put> puts = new ArrayList<>();
for (HRow hRow : rows) {
Put put = new Put(hRow.getRowKey());
@@ -145,7 +159,7 @@ public class HbaseTemplate {
public Boolean deletes(String tableName, Set<byte[]> rowKeys) {
boolean flag = false;
try {
HTable table = (HTable) conn.getTable(TableName.valueOf(tableName));
HTable table = (HTable) getConnection().getTable(TableName.valueOf(tableName));
List<Delete> deletes = new ArrayList<>();
for (byte[] rowKey : rowKeys) {
Delete delete = new Delete(rowKey);
@@ -160,4 +174,10 @@ public class HbaseTemplate {
}
return flag;
}
public void close() throws IOException {
if (conn != null) {
conn.close();
}
}
}

49
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/service/RdbSyncService.java
View File

@@ -7,10 +7,7 @@ import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.*;
import java.util.function.Function;
import javax.sql.DataSource;
@@ -88,7 +85,6 @@ public class RdbSyncService {
* @param function 回调方法
*/
public void sync(List<Dml> dmls, Function<Dml, Boolean> function) {
try {
boolean toExecute = false;
for (Dml dml : dmls) {
if (!toExecute) {
@@ -102,27 +98,26 @@ public class RdbSyncService {
for (int i = 0; i < threads; i++) {
int j = i;
futures.add(executorThreads[i].submit(() -> {
try {
dmlsPartition[j]
.forEach(syncItem -> sync(batchExecutors[j], syncItem.config, syncItem.singleDml));
dmlsPartition[j].clear();
batchExecutors[j].commit();
return true;
} catch (Exception e) {
batchExecutors[j].rollback();
throw new RuntimeException(e);
}
}));
}
futures.forEach(future -> {
try {
future.get();
} catch (Exception e) {
logger.error(e.getMessage(), e);
} catch (ExecutionException | InterruptedException e) {
throw new RuntimeException(e);
}
});
for (int i = 0; i < threads; i++) {
dmlsPartition[i].clear();
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
@@ -181,7 +176,6 @@ public class RdbSyncService {
* @param dml DML
*/
public void sync(BatchExecutor batchExecutor, MappingConfig config, SingleDml dml) {
try {
if (config != null) {
String type = dml.getType();
if (type != null && type.equalsIgnoreCase("INSERT")) {
@@ -195,9 +189,6 @@ public class RdbSyncService {
logger.debug("DML: {}", JSON.toJSONString(dml, SerializerFeature.WriteMapNullValue));
}
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
/**
@@ -214,7 +205,6 @@ public class RdbSyncService {
DbMapping dbMapping = config.getDbMapping();
try {
Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, data);
StringBuilder insertSql = new StringBuilder();
@@ -241,7 +231,9 @@ public class RdbSyncService {
}
Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
if (type == null) {
throw new RuntimeException("Target column: " + targetColumnName + " not matched");
}
Object value = data.get(srcColumnName);
BatchExecutor.setValue(values, type, value);
@@ -252,9 +244,6 @@ public class RdbSyncService {
logger.trace("Insert into target table, sql: {}", insertSql);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
/**
@@ -276,7 +265,6 @@ public class RdbSyncService {
DbMapping dbMapping = config.getDbMapping();
try {
Map<String, String> columnsMap = SyncUtil.getColumnsMap(dbMapping, data);
Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
@@ -296,6 +284,9 @@ public class RdbSyncService {
for (String targetColumnName : targetColumnNames) {
updateSql.append(targetColumnName).append("=?, ");
Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
if (type == null) {
throw new RuntimeException("Target column: " + targetColumnName + " not matched");
}
BatchExecutor.setValue(values, type, data.get(srcColumnName));
}
}
@@ -311,9 +302,6 @@ public class RdbSyncService {
if (logger.isTraceEnabled()) {
logger.trace("Update target table, sql: {}", updateSql);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
/**
@@ -330,7 +318,6 @@ public class RdbSyncService {
DbMapping dbMapping = config.getDbMapping();
try {
Map<String, Integer> ctype = getTargetColumnType(batchExecutor.getConn(), config);
StringBuilder sql = new StringBuilder();
@@ -345,9 +332,6 @@ public class RdbSyncService {
if (logger.isTraceEnabled()) {
logger.trace("Delete from target table, sql: {}", sql);
}
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
}
/**
@@ -405,6 +389,9 @@ public class RdbSyncService {
}
sql.append(targetColumnName).append("=? AND ");
Integer type = ctype.get(Util.cleanColumn(targetColumnName).toLowerCase());
if (type == null) {
throw new RuntimeException("Target column: " + targetColumnName + " not matched");
}
// 如果有修改主键的情况
if (o != null && o.containsKey(srcColumnName)) {
BatchExecutor.setValue(values, type, o.get(srcColumnName));

24
client-adapter/rdb/src/main/java/com/alibaba/otter/canal/client/adapter/rdb/support/BatchExecutor.java
View File

@@ -16,18 +16,11 @@ public class BatchExecutor implements Closeable {
private static final Logger logger = LoggerFactory.getLogger(BatchExecutor.class);
private Integer key;
private Connection conn;
private AtomicInteger idx = new AtomicInteger(0);
public BatchExecutor(Connection conn){
this(1, conn);
}
public BatchExecutor(Integer key, Connection conn){
this.key = key;
this.conn = conn;
try {
this.conn.setAutoCommit(false);
} catch (SQLException e) {
@@ -35,9 +28,6 @@ public class BatchExecutor implements Closeable {
}
}
public Integer getKey() {
return key;
}
public Connection getConn() {
return conn;
@@ -70,7 +60,19 @@ public class BatchExecutor implements Closeable {
try {
conn.commit();
if (logger.isTraceEnabled()) {
logger.trace("Batch executor: " + key + " commit " + idx.get() + " rows");
logger.trace("Batch executor commit " + idx.get() + " rows");
}
idx.set(0);
} catch (SQLException e) {
logger.error(e.getMessage(), e);
}
}
public void rollback() {
try {
conn.rollback();
if (logger.isTraceEnabled()) {
logger.trace("Batch executor rollback " + idx.get() + " rows");
}
idx.set(0);
} catch (SQLException e) {

14
client-adapter/rdb/src/main/resources/rdb/mytest_user.yml
View File

@@ -8,13 +8,13 @@ dbMapping:
targetTable: mytest.tb_user
targetPk:
id: id
mapAll: true
# targetColumns:
# id:
# name:
# role_id:
# c_time:
# test1:
# mapAll: true
targetColumns:
id:
name:
role_id:
c_time:
test1:
# Mirror schema synchronize config