From b5697a366abcf0c03e5a58d879740ac0f196dd71 Mon Sep 17 00:00:00 2001
From: mcy <rewerma@163.com>
Date: Sun, 23 Dec 2018 18:24:04 +0800
Subject: [PATCH 1/2] merge

---
 .../client/kafka/KafkaCanalConnector.java      | 18 +++++++++---------
 1 file changed, 9 insertions(+), 9 deletions(-)

diff --git a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
index 5e9c99c2..9b8b263a 100644
--- a/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
+++ b/client/src/main/java/com/alibaba/otter/canal/client/kafka/KafkaCanalConnector.java
@@ -30,16 +30,16 @@ import com.google.common.collect.Lists;
  */
 public class KafkaCanalConnector implements CanalMQConnector {
 
-    private KafkaConsumer<String, Message> kafkaConsumer;
-    private KafkaConsumer<String, String>  kafkaConsumer2;                   // 用于扁平message的数据消费
-    private String                         topic;
-    private Integer                        partition;
-    private Properties                     properties;
-    private volatile boolean               connected       = false;
-    private volatile boolean               running         = false;
-    private boolean                        flatMessage;
+    protected KafkaConsumer<String, Message> kafkaConsumer;
+    protected KafkaConsumer<String, String>  kafkaConsumer2;                  // 用于扁平message的数据消费
+    protected String                         topic;
+    protected Integer                        partition;
+    protected Properties                     properties;
+    protected volatile boolean               connected      = false;
+    protected volatile boolean               running        = false;
+    protected boolean                        flatMessage;
 
-    private Map<Integer, Long>             currentOffsets  = new HashMap<>();
+    private Map<Integer, Long>               currentOffsets = new HashMap<>();
 
     public KafkaCanalConnector(String servers, String topic, Integer partition, String groupId, Integer batchSize,
                                boolean flatMessage){

From dffbc437a6f9204c1a3bf6b4304ea45e6aa3fa62 Mon Sep 17 00:00:00 2001
From: machey <rewerma@163.com>
Date: Tue, 25 Dec 2018 22:27:27 +0800
Subject: [PATCH 2/2] fixed #1338

---
 .../canal/client/adapter/es/ESAdapter.java    | 22 +++++++++----------
 .../client/adapter/hbase/HbaseAdapter.java    |  4 +++-
 2 files changed, 13 insertions(+), 13 deletions(-)

diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
index 9bb56792..dd82cafc 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/ESAdapter.java
@@ -75,8 +75,8 @@ public class ESAdapter implements OuterAdapter {
             // 过滤不匹配的key的配置
             esSyncConfigTmp.forEach((key, config) -> {
                 if ((config.getOuterAdapterKey() == null && configuration.getKey() == null)
-                    || (config.getOuterAdapterKey() != null && config.getOuterAdapterKey()
-                        .equalsIgnoreCase(configuration.getKey()))) {
+                    || (config.getOuterAdapterKey() != null
+                        && config.getOuterAdapterKey().equalsIgnoreCase(configuration.getKey()))) {
                     esSyncConfig.put(key, config);
                 }
             });
@@ -98,15 +98,11 @@ public class ESAdapter implements OuterAdapter {
                 }
                 String schema = matcher.group(2);
 
-                schemaItem.getAliasTableItems()
-                    .values()
-                    .forEach(tableItem -> {
-                        Map<String, ESSyncConfig> esSyncConfigMap = dbTableEsSyncConfig.computeIfAbsent(schema
-                                                                                                        + "-"
-                                                                                                        + tableItem.getTableName(),
-                            k -> new HashMap<>());
-                        esSyncConfigMap.put(configName, config);
-                    });
+                schemaItem.getAliasTableItems().values().forEach(tableItem -> {
+                    Map<String, ESSyncConfig> esSyncConfigMap = dbTableEsSyncConfig
+                        .computeIfAbsent(schema + "-" + tableItem.getTableName(), k -> new HashMap<>());
+                    esSyncConfigMap.put(configName, config);
+                });
             }
 
             Map<String, String> properties = configuration.getProperties();
@@ -140,7 +136,9 @@ public class ESAdapter implements OuterAdapter {
         String database = dml.getDatabase();
         String table = dml.getTable();
         Map<String, ESSyncConfig> configMap = dbTableEsSyncConfig.get(database + "-" + table);
-        esSyncService.sync(configMap.values(), dml);
+        if (configMap != null) {
+            esSyncService.sync(configMap.values(), dml);
+        }
     }
 
     @Override
diff --git a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
index e586bfe0..7377c43e 100644
--- a/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
+++ b/client-adapter/hbase/src/main/java/com/alibaba/otter/canal/client/adapter/hbase/HbaseAdapter.java
@@ -115,7 +115,9 @@ public class HbaseAdapter implements OuterAdapter {
         String database = dml.getDatabase();
         String table = dml.getTable();
         Map<String, MappingConfig> configMap = mappingConfigCache.get(destination + "." + database + "." + table);
-        configMap.values().forEach(config -> hbaseSyncService.sync(config, dml));
+        if (configMap != null) {
+            configMap.values().forEach(config -> hbaseSyncService.sync(config, dml));
+        }
     }
 
     @Override