Make source compatible with change jprante/elasticsearch-jdbc#655 (still unmerged so compilation will fail).

Author: feaster83

Diff for: pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.feaster83.elasticsearch-river-jdbc</groupId>
     <artifactId>rebuild_strategy</artifactId>
-    <version>1.7.0.1-SNAPSHOT</version>
+    <version>1.7.2.0-SNAPSHOT</version>
     <packaging>jar</packaging>
 
     <developers>
@@ -52,7 +52,7 @@
         <dependency>
             <groupId>org.xbib.elasticsearch.importer</groupId>
             <artifactId>elasticsearch-jdbc</artifactId>
-            <version>1.7.0.1</version>
+            <version>1.7.2.0</version>
         </dependency>
         <dependency>
             <groupId>junit</groupId>
@@ -62,4 +62,4 @@
         </dependency>
     </dependencies>
 
-</project>
+</project>

Diff for: src/main/java/org/xbib/elasticsearch/jdbc/strategy/standard/RebuildSink.java

@@ -27,6 +27,7 @@
 import org.elasticsearch.common.lang3.StringUtils;
 import org.elasticsearch.common.unit.TimeValue;
 import org.xbib.elasticsearch.common.util.IndexableObject;
+import org.xbib.elasticsearch.support.client.Ingest;
 
 import java.io.File;
 import java.io.IOException;
@@ -98,14 +99,8 @@ public RebuildSink<C> setContext(StandardContext context) {
 
     @Override
     public synchronized void beforeFetch() throws IOException {
-        if (ingest == null) {
-            if (context.getIngestFactory() != null) {
-                ingest = context.getIngestFactory().create();
-                ingest.setMetric(getMetric());
-            } else {
-                logger.warn("no ingest factory found");
-            }
-        }
+        Ingest ingest = context.getOrCreateIngest(getMetric());
+
         if (ingest == null) {
             logger.warn("no ingest found");
             return;
@@ -120,6 +115,7 @@ public synchronized void beforeFetch() throws IOException {
     }
 
     private void createIndex() throws IOException {
+        Ingest ingest = context.getOrCreateIngest(getMetric());
         rebuild_index = generateNewIndexName();
         index = rebuild_index;
 
@@ -153,11 +149,11 @@ private String generateNewIndexName() {
 
     @Override
     public synchronized void afterFetch() throws IOException {
-        if (ingest == null) {
-            ingest = context.getIngestFactory().create();
-        }
+        Ingest ingest = context.getIngest();
 
-        flushIngest();
+        if(ingest == null) {
+            return;
+        }
 
         ingest.stopBulk(rebuild_index);
         ingest.refreshIndex(rebuild_index);
@@ -177,11 +173,6 @@ public synchronized void afterFetch() throws IOException {
         allExistingIndices.remove(rebuild_index);
 
         removeOldIndices(allExistingIndices);
-
-        if (ingest != null) {
-            ingest.shutdown();
-        }
-        ingest = null;
     }
 
     @Override
@@ -218,6 +209,7 @@ public int compare(DateTime o1, DateTime o2) {
     }
 
     private List<String> getAllExistingIndices() {
+        Ingest ingest = context.getIngest();
         try {
             GetIndexResponse indexResponse = ingest.client().admin().indices().prepareGetIndex().addIndices(index_prefix + "*").execute().get();
 
@@ -237,6 +229,7 @@ private List<String> getAllExistingIndices() {
     }
 
     private List<String> getCurrentActiveIndices() {
+        Ingest ingest = context.getIngest();
         GetAliasesResponse getAliasesResponse = ingest.client().admin().indices().prepareGetAliases(alias).execute().actionGet();
 
         List<String> existingIndices = new ArrayList<>();
@@ -250,6 +243,7 @@ private List<String> getCurrentActiveIndices() {
     }
 
     private void removeOldIndices(List<String> indices) {
+        Ingest ingest = context.getIngest();
         List<String> orderedIndexList = getOrderedIndexList(indices);
         for (int i = 0; i < (orderedIndexList.size() - keep_last_indices); i++) {
             String existingIndex = orderedIndexList.get(i);
@@ -259,6 +253,7 @@ private void removeOldIndices(List<String> indices) {
     }
 
     private void switchAliasToNewIndex(List<String> existingIndices) {
+        Ingest ingest = context.getIngest();
         IndicesAliasesRequestBuilder prepareAliasesRequest = ingest.client().admin().indices().prepareAliases();
         prepareAliasesRequest.addAlias(rebuild_index, alias).execute().actionGet().isAcknowledged();