+3
-2
@@ -36,6 +36,7 @@ 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;
|
||||
|
||||
|
||||
/**
|
||||
* ES外部适配器
|
||||
*
|
||||
@@ -75,7 +76,6 @@ public class ESAdapter implements OuterAdapter {
|
||||
@Override
|
||||
public void init(OuterAdapterConfig configuration, Properties envProperties) {
|
||||
try {
|
||||
MDC.put("adapter", "es");
|
||||
this.envProperties = envProperties;
|
||||
Map<String, ESSyncConfig> esSyncConfigTmp = ESSyncConfigLoader.load(envProperties);
|
||||
// 过滤不匹配的key的配置
|
||||
@@ -145,6 +145,7 @@ public class ESAdapter implements OuterAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sync(List<Dml> dmls) {
|
||||
if (dmls == null || dmls.isEmpty()) {
|
||||
return;
|
||||
@@ -155,6 +156,7 @@ public class ESAdapter implements OuterAdapter {
|
||||
}
|
||||
}
|
||||
esSyncService.commit(); // 批次统一提交
|
||||
|
||||
}
|
||||
|
||||
private void sync(Dml dml) {
|
||||
@@ -245,7 +247,6 @@ public class ESAdapter implements OuterAdapter {
|
||||
if (transportClient != null) {
|
||||
transportClient.close();
|
||||
}
|
||||
MDC.remove("adapter");
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+2
-4
@@ -26,7 +26,6 @@ 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.*;
|
||||
import org.slf4j.MDC;
|
||||
|
||||
/**
|
||||
* HBase外部适配器
|
||||
@@ -60,7 +59,6 @@ public class HbaseAdapter implements OuterAdapter {
|
||||
@Override
|
||||
public void init(OuterAdapterConfig configuration, Properties envProperties) {
|
||||
try {
|
||||
MDC.put("adapter", "hbase");
|
||||
this.envProperties = envProperties;
|
||||
Map<String, MappingConfig> hbaseMappingTmp = MappingConfigLoader.load(envProperties);
|
||||
// 过滤不匹配的key的配置
|
||||
@@ -104,6 +102,7 @@ public class HbaseAdapter implements OuterAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public void sync(List<Dml> dmls) {
|
||||
if (dmls == null || dmls.isEmpty()) {
|
||||
return;
|
||||
@@ -113,7 +112,7 @@ public class HbaseAdapter implements OuterAdapter {
|
||||
}
|
||||
}
|
||||
|
||||
public void sync(Dml dml) {
|
||||
private void sync(Dml dml) {
|
||||
if (dml == null) {
|
||||
return;
|
||||
}
|
||||
@@ -224,7 +223,6 @@ public class HbaseAdapter implements OuterAdapter {
|
||||
} catch (IOException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
MDC.remove("adapter");
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
@@ -40,11 +40,11 @@ if [ -z "$JAVA" ]; then
|
||||
fi
|
||||
fi
|
||||
|
||||
case "$#"
|
||||
case "$#"
|
||||
in
|
||||
0 )
|
||||
0 )
|
||||
;;
|
||||
2 )
|
||||
2 )
|
||||
if [ "$1" = "debug" ]; then
|
||||
DEBUG_PORT=$2
|
||||
DEBUG_SUSPEND="n"
|
||||
@@ -76,8 +76,8 @@ echo "cd to $bin_abs_path for workaround relative path"
|
||||
cd $bin_abs_path
|
||||
|
||||
echo CLASSPATH :$CLASSPATH
|
||||
$JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $ADAPTER_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication 1>>$base/logs/adapter.log 2>&1 &
|
||||
$JAVA $JAVA_OPTS $JAVA_DEBUG_OPT $ADAPTER_OPTS -classpath .:$CLASSPATH com.alibaba.otter.canal.adapter.launcher.CanalAdapterApplication 1>>/dev/null 2>&1 &
|
||||
echo $! > $base/bin/adapter.pid
|
||||
|
||||
echo "cd to $current_path for continue"
|
||||
cd $current_path
|
||||
cd $current_path
|
||||
|
||||
@@ -8,17 +8,17 @@
|
||||
</appender>
|
||||
|
||||
<appender name="CANAL-ROOT" class="ch.qos.logback.classic.sift.SiftingAppender">
|
||||
<discriminator>
|
||||
<Key>adapter</Key>
|
||||
<DefaultValue>adapter</DefaultValue>
|
||||
</discriminator>
|
||||
<discriminator>
|
||||
<Key>adapter</Key>
|
||||
<DefaultValue>adapter</DefaultValue>
|
||||
</discriminator>
|
||||
<sift>
|
||||
<appender name="FILE-${adapter}" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||
<File>../logs/${adapter}/${adapter}.log</File>
|
||||
<appender name="FILE-adapter" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||
<File>../logs/adapter/adapter.log</File>
|
||||
<rollingPolicy
|
||||
class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
|
||||
<!-- rollover daily -->
|
||||
<fileNamePattern>../logs/${adapter}/%d{yyyy-MM-dd}/${adapter}-%d{yyyy-MM-dd}-%i.log.gz</fileNamePattern>
|
||||
<fileNamePattern>../logs/adapter/%d{yyyy-MM-dd}/adapter-%d{yyyy-MM-dd}-%i.log.gz</fileNamePattern>
|
||||
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
|
||||
<!-- or whenever the file size reaches 100MB -->
|
||||
<maxFileSize>512MB</maxFileSize>
|
||||
@@ -36,14 +36,22 @@
|
||||
|
||||
<logger name="com.alibaba.otter.canal.client.adapter.hbase" additivity="false">
|
||||
<level value="DEBUG" />
|
||||
<appender-ref ref="STDOUT"/>
|
||||
<appender-ref ref="CANAL-ROOT" />
|
||||
</logger>
|
||||
<logger name="com.alibaba.otter.canal.client.adapter.es" additivity="false">
|
||||
<level value="DEBUG" />
|
||||
<appender-ref ref="STDOUT"/>
|
||||
<appender-ref ref="CANAL-ROOT" />
|
||||
</logger>
|
||||
<logger name="com.alibaba.otter.canal.client.adapter.rdb" additivity="false">
|
||||
<level value="DEBUG" />
|
||||
<appender-ref ref="STDOUT"/>
|
||||
<appender-ref ref="CANAL-ROOT" />
|
||||
</logger>
|
||||
<logger name="com.alibaba.otter.canal.client.adapter.logger" additivity="false">
|
||||
<level value="DEBUG" />
|
||||
<appender-ref ref="STDOUT"/>
|
||||
<appender-ref ref="CANAL-ROOT" />
|
||||
</logger>
|
||||
|
||||
|
||||
-3
@@ -5,7 +5,6 @@ import java.util.Properties;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.slf4j.MDC;
|
||||
|
||||
import com.alibaba.fastjson.JSON;
|
||||
import com.alibaba.fastjson.serializer.SerializerFeature;
|
||||
@@ -28,7 +27,6 @@ public class LoggerAdapterExample implements OuterAdapter {
|
||||
|
||||
@Override
|
||||
public void init(OuterAdapterConfig configuration, Properties envProperties) {
|
||||
MDC.put("adapter", "logger");
|
||||
}
|
||||
|
||||
public void sync(List<Dml> dmls) {
|
||||
@@ -43,6 +41,5 @@ public class LoggerAdapterExample implements OuterAdapter {
|
||||
|
||||
@Override
|
||||
public void destroy() {
|
||||
MDC.remove("adapter");
|
||||
}
|
||||
}
|
||||
|
||||
+5
-13
@@ -14,7 +14,6 @@ import org.apache.commons.lang.BooleanUtils;
|
||||
import org.apache.commons.lang.StringUtils;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.slf4j.MDC;
|
||||
|
||||
import com.alibaba.druid.pool.DruidDataSource;
|
||||
import com.alibaba.otter.canal.client.adapter.OuterAdapter;
|
||||
@@ -26,12 +25,7 @@ import com.alibaba.otter.canal.client.adapter.rdb.service.RdbEtlService;
|
||||
import com.alibaba.otter.canal.client.adapter.rdb.service.RdbMirrorDbSyncService;
|
||||
import com.alibaba.otter.canal.client.adapter.rdb.service.RdbSyncService;
|
||||
import com.alibaba.otter.canal.client.adapter.rdb.support.SyncUtil;
|
||||
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.Util;
|
||||
import com.alibaba.otter.canal.client.adapter.support.*;
|
||||
|
||||
/**
|
||||
* RDB适配器实现类
|
||||
@@ -76,14 +70,13 @@ public class RdbAdapter implements OuterAdapter {
|
||||
*/
|
||||
@Override
|
||||
public void init(OuterAdapterConfig configuration, Properties envProperties) {
|
||||
MDC.put("adapter", "rdb");
|
||||
this.envProperties = envProperties;
|
||||
Map<String, MappingConfig> rdbMappingTmp = ConfigLoader.load(envProperties);
|
||||
// 过滤不匹配的key的配置
|
||||
rdbMappingTmp.forEach((key, mappingConfig) -> {
|
||||
if ((mappingConfig.getOuterAdapterKey() == null && configuration.getKey() == null)
|
||||
|| (mappingConfig.getOuterAdapterKey() != null && mappingConfig.getOuterAdapterKey()
|
||||
.equalsIgnoreCase(configuration.getKey()))) {
|
||||
|| (mappingConfig.getOuterAdapterKey() != null
|
||||
&& mappingConfig.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
|
||||
rdbMapping.put(key, mappingConfig);
|
||||
}
|
||||
});
|
||||
@@ -143,8 +136,8 @@ public class RdbAdapter implements OuterAdapter {
|
||||
String threads = properties.get("threads");
|
||||
// String commitSize = properties.get("commitSize");
|
||||
|
||||
boolean skipDupException = BooleanUtils.toBoolean(configuration.getProperties()
|
||||
.getOrDefault("skipDupException", "true"));
|
||||
boolean skipDupException = BooleanUtils
|
||||
.toBoolean(configuration.getProperties().getOrDefault("skipDupException", "true"));
|
||||
rdbSyncService = new RdbSyncService(dataSource,
|
||||
threads != null ? Integer.valueOf(threads) : null,
|
||||
skipDupException);
|
||||
@@ -304,6 +297,5 @@ public class RdbAdapter implements OuterAdapter {
|
||||
if (dataSource != null) {
|
||||
dataSource.close();
|
||||
}
|
||||
MDC.remove("adapter");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user