From 17efc2ada01b677c09d26db15051c0ffce9b83d4 Mon Sep 17 00:00:00 2001
From: mcy <rewerma@163.com>
Date: Fri, 22 Feb 2019 16:56:58 +0800
Subject: [PATCH] =?UTF-8?q?es=20=E6=89=B9=E9=87=8F=E6=8F=90=E4=BA=A4?=
 =?UTF-8?q?=E4=BF=AE=E6=94=B9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../canal/client/adapter/es/support/ESTemplate.java    | 10 +++++++---
 1 file changed, 7 insertions(+), 3 deletions(-)

diff --git a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
index 5974883d..997f1366 100644
--- a/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
+++ b/client-adapter/elasticsearch/src/main/java/com/alibaba/otter/canal/client/adapter/es/support/ESTemplate.java
@@ -72,6 +72,7 @@ public class ESTemplate {
                 getBulk().add(transportClient.prepareIndex(mapping.get_index(), mapping.get_type(), pkVal.toString())
                     .setSource(esFieldData));
             }
+            commitBulk();
         } else {
             SearchResponse response = transportClient.prepareSearch(mapping.get_index())
                 .setTypes(mapping.get_type())
@@ -81,9 +82,10 @@ public class ESTemplate {
             for (SearchHit hit : response.getHits()) {
                 getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId())
                     .setDoc(esFieldData));
+                commitBulk();
             }
         }
-        commitBulk();
+
     }
 
     /**
@@ -149,6 +151,7 @@ public class ESTemplate {
     public void delete(ESMapping mapping, Object pkVal, Map<String, Object> esFieldData) {
         if (mapping.get_id() != null) {
             getBulk().add(transportClient.prepareDelete(mapping.get_index(), mapping.get_type(), pkVal.toString()));
+            commitBulk();
         } else {
             SearchResponse response = transportClient.prepareSearch(mapping.get_index())
                 .setTypes(mapping.get_type())
@@ -158,9 +161,10 @@ public class ESTemplate {
             for (SearchHit hit : response.getHits()) {
                 getBulk().add(transportClient.prepareUpdate(mapping.get_index(), mapping.get_type(), hit.getId())
                     .setDoc(esFieldData));
+                commitBulk();
             }
         }
-        commitBulk();
+
     }
 
     /**
@@ -186,7 +190,7 @@ public class ESTemplate {
     }
 
     /**
-     * 如果大于批量则提交批次
+     * 如果大于批量数则提交批次
      */
     private void commitBulk() {
         if (getBulk().numberOfActions() >= MAX_BATCH_SIZE) {