Merge pull request #2 from aiven/add-es7-support

Add ES7 support

Author: fingon

pom.xml

@@ -58,7 +58,7 @@
         <kafka.version>2.2.0</kafka.version>
         <es.version>2.4.1</es.version>
         <lucene.version>5.5.2</lucene.version>
-        <jest.version>2.4.0</jest.version>
+        <jest.version>6.3.1</jest.version>
 
         <junit.version>4.12</junit.version>
         <hamcrest.version>1.3</hamcrest.version>

src/main/java/io/aiven/connect/elasticsearch/DataConverter.java

@@ -321,7 +321,7 @@ private Object preProcessLogicalValue(String schemaName, Object value) {
   }
 
   private Object preProcessArrayValue(Object value, Schema schema, Schema newSchema) {
-    Collection collection = (Collection) value;
+    Collection<?> collection = (Collection<?>) value;
     List<Object> result = new ArrayList<>();
     for (Object element: collection) {
       result.add(preProcessValue(element, schema.valueSchema(), newSchema.valueSchema()));

src/main/java/io/aiven/connect/elasticsearch/ElasticsearchClient.java

@@ -29,7 +29,7 @@
 public interface ElasticsearchClient extends AutoCloseable {
 
   enum Version {
-    ES_V1, ES_V2, ES_V5, ES_V6
+    ES_V1, ES_V2, ES_V5, ES_V6, ES_V7
   }
 
   /**
@@ -96,5 +96,5 @@ enum Version {
   /**
    * Shuts down the client.
    */
-  void close();
+  void close() throws IOException;
 }

src/main/java/io/aiven/connect/elasticsearch/ElasticsearchSinkTask.java

@@ -28,6 +28,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.io.IOException;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -182,7 +183,11 @@ public void stop() throws ConnectException {
       writer.stop();
     }
     if (client != null) {
-      client.close();
+      try {
+        client.close();
+      } catch (IOException e) {
+        throw new ConnectException(e);
+      }
     }
   }
 

src/main/java/io/aiven/connect/elasticsearch/jest/JestElasticsearchClient.java

@@ -198,6 +198,8 @@ private Version getServerVersion() throws IOException {
       return Version.ES_V5;
     } else if (esVersion.startsWith("6.")) {
       return Version.ES_V6;
+    } else if (esVersion.startsWith("7.")) {
+      return Version.ES_V7;
     }
     return defaultVersion;
   }
@@ -229,7 +231,11 @@ private boolean indexExists(String index) {
   public void createIndices(Set<String> indices) {
     for (String index : indices) {
       if (!indexExists(index)) {
-        CreateIndex createIndex = new CreateIndex.Builder(index).build();
+        CreateIndex.Builder builder = new CreateIndex.Builder(index);
+        if (version.equals(Version.ES_V7)) {
+          builder.setParameter("include_type_name", true);
+        }
+        CreateIndex createIndex = builder.build();
         try {
           JestResult result = client.execute(createIndex);
           if (!result.isSucceeded()) {
@@ -285,7 +291,7 @@ public BulkRequest createBulkRequest(List<IndexableRecord> batch) {
   }
 
   // visible for testing
-  protected BulkableAction toBulkableAction(IndexableRecord record) {
+  protected BulkableAction<?> toBulkableAction(IndexableRecord record) {
     // If payload is null, the record was a tombstone and we should delete from the index.
     return record.payload != null ? toIndexRequest(record) : toDeleteRequest(record);
   }
@@ -364,7 +370,7 @@ public JsonObject search(String query, String index, String type) throws IOExcep
     return result.getJsonObject();
   }
 
-  public void close() {
-    client.shutdownClient();
+  public void close() throws IOException {
+    client.close();
   }
 }

src/test/java/io/aiven/connect/elasticsearch/DataConverterTest.java

@@ -234,7 +234,7 @@ public void stringKeyedMapCompactFormat() {
         SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.INT32_SCHEMA).build(),
         preProcessedSchema
     );
-    HashMap newValue = (HashMap) converter.preProcessValue(origValue, origSchema, preProcessedSchema);
+    HashMap<?, ?> newValue = (HashMap<?, ?>) converter.preProcessValue(origValue, origSchema, preProcessedSchema);
     assertEquals(origValue, newValue);
   }
 

src/test/java/io/aiven/connect/elasticsearch/bulk/BulkProcessorTest.java

@@ -375,10 +375,10 @@ public void ignoreOrWarnOnMalformedDoc() throws InterruptedException {
     // Test both IGNORE and WARN options
     // There is no difference in logic between IGNORE and WARN, except for the logging.
     // Test to ensure they both work the same logically
-    final List<BehaviorOnMalformedDoc> behaviorsToTest = new ArrayList<BehaviorOnMalformedDoc>() {{
-      add(BehaviorOnMalformedDoc.WARN);
-      add(BehaviorOnMalformedDoc.IGNORE);
-    }};
+    final List<BehaviorOnMalformedDoc> behaviorsToTest = Arrays.asList(
+        BehaviorOnMalformedDoc.WARN,
+        BehaviorOnMalformedDoc.IGNORE
+    );
 
     for(BehaviorOnMalformedDoc behaviorOnMalformedDoc : behaviorsToTest)
     {

src/test/java/io/aiven/connect/elasticsearch/jest/JestElasticsearchClientTest.java

@@ -31,6 +31,7 @@
 import io.searchbox.client.JestClient;
 import io.searchbox.client.JestClientFactory;
 import io.searchbox.client.JestResult;
+import io.searchbox.client.config.ElasticsearchVersion;
 import io.searchbox.client.config.HttpClientConfig;
 import io.searchbox.cluster.NodesInfo;
 import io.searchbox.core.BulkResult;
@@ -52,6 +53,7 @@
 import org.mockito.ArgumentMatcher;
 import org.mockito.InOrder;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -169,7 +171,7 @@ private ArgumentMatcher<CreateIndex> isCreateIndexForTestIndex() {
       @Override
       public boolean matches(CreateIndex createIndex) {
         // check the URI as the equals method on CreateIndex doesn't work
-        return createIndex.getURI().equals(INDEX);
+        return createIndex.getURI(ElasticsearchVersion.V2).equals(INDEX);
       }
     };
   }
@@ -313,11 +315,11 @@ public void searches() throws Exception {
   }
 
   @Test
-  public void closes() {
+  public void closes() throws IOException {
     JestElasticsearchClient client = new JestElasticsearchClient(jestClient);
     client.close();
 
-    verify(jestClient).shutdownClient();
+    verify(jestClient).close();
   }
 
   private BulkResult createBulkResultFailure(String exception) {