* Apply general review feedback on naming and conventions

* Update indexes to indices per comment by @rayokota
* Introduce a local in memory cache to prevent recurrent calls to indexExist to Elasticsearch
* use a HashSet as a cache and not a HashMap
* general code style

Author: purbon

src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfig.java

@@ -139,9 +139,9 @@ public class ElasticsearchSinkConnectorConfig extends AbstractConfig {
       + " mapping conflict or a field name containing illegal characters. Valid options are "
       + "'ignore', 'warn', and 'fail'.";
 
-  public static final String CREATE_INDEX_AT_START_CONFIG = "create.index.at.start";
-  private static final String CREATE_INDEX_AT_START__DOC = "Create the Elasticsearch indexes at "
-      + " bootstrap time. This is useful when the indexes are a direct mapping "
+  public static final String AUTO_CREATE_INDICES_AT_START_CONFIG = "auto.create.indices.at.start";
+  private static final String AUTO_CREATE_INDICES_AT_START_DOC = "Auto create the Elasticsearch"
+      + " indices at bootstrap time. This is useful when the indices are a direct mapping "
       + " of the Kafka topics.";
 
   protected static ConfigDef baseConfigDef() {
@@ -254,15 +254,16 @@ private static void addConnectorConfigs(ConfigDef configDef) {
         Width.SHORT,
         "Read Timeout"
     ).define(
-        CREATE_INDEX_AT_START_CONFIG,
+        AUTO_CREATE_INDICES_AT_START_CONFIG,
         Type.BOOLEAN,
         true,
         Importance.LOW,
-        CREATE_INDEX_AT_START__DOC,
+        AUTO_CREATE_INDICES_AT_START_DOC,
         group,
         ++order,
         Width.SHORT,
-        "Create indexes at bootstrap time");
+        "Create indices at bootstrap time"
+      );
   }
 
   private static void addConversionConfigs(ConfigDef configDef) {

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

@@ -40,7 +40,7 @@ public class ElasticsearchSinkTask extends SinkTask {
   private static final Logger log = LoggerFactory.getLogger(ElasticsearchSinkTask.class);
   private ElasticsearchWriter writer;
   private ElasticsearchClient client;
-  private Boolean createIndexAtStartTime;
+  private Boolean createIndicesAtStartTime;
 
   @Override
   public String version() {
@@ -91,8 +91,8 @@ public void start(Map<String, String> props, ElasticsearchClient client) {
           config.getInt(ElasticsearchSinkConnectorConfig.MAX_RETRIES_CONFIG);
       boolean dropInvalidMessage =
           config.getBoolean(ElasticsearchSinkConnectorConfig.DROP_INVALID_MESSAGE_CONFIG);
-      boolean createIndexAtStartTime =
-          config.getBoolean(ElasticsearchSinkConnectorConfig.CREATE_INDEX_AT_START_CONFIG);
+      boolean createIndicesAtStartTime =
+          config.getBoolean(ElasticsearchSinkConnectorConfig.AUTO_CREATE_INDICES_AT_START_CONFIG);
 
       DataConverter.BehaviorOnNullValues behaviorOnNullValues =
           DataConverter.BehaviorOnNullValues.forValue(
@@ -139,7 +139,7 @@ public void start(Map<String, String> props, ElasticsearchClient client) {
           .setBehaviorOnNullValues(behaviorOnNullValues)
           .setBehaviorOnMalformedDoc(behaviorOnMalformedDoc);
 
-      this.createIndexAtStartTime = createIndexAtStartTime;
+      this.createIndicesAtStartTime = createIndicesAtStartTime;
 
       writer = builder.build();
       writer.start();
@@ -154,7 +154,7 @@ public void start(Map<String, String> props, ElasticsearchClient client) {
   @Override
   public void open(Collection<TopicPartition> partitions) {
     log.debug("Opening the task for topic partitions: {}", partitions);
-    if (createIndexAtStartTime) {
+    if (createIndicesAtStartTime) {
       Set<String> topics = new HashSet<>();
       for (TopicPartition tp : partitions) {
         topics.add(tp.topic());

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

@@ -20,12 +20,12 @@
 import com.fasterxml.jackson.databind.node.JsonNodeFactory;
 import com.fasterxml.jackson.databind.node.ObjectNode;
 import com.google.gson.JsonObject;
-import io.confluent.connect.elasticsearch.bulk.BulkRequest;
 import io.confluent.connect.elasticsearch.ElasticsearchClient;
 import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig;
 import io.confluent.connect.elasticsearch.IndexableRecord;
 import io.confluent.connect.elasticsearch.Key;
 import io.confluent.connect.elasticsearch.Mapping;
+import io.confluent.connect.elasticsearch.bulk.BulkRequest;
 import io.confluent.connect.elasticsearch.bulk.BulkResponse;
 import io.searchbox.action.Action;
 import io.searchbox.action.BulkableAction;
@@ -44,17 +44,17 @@
 import io.searchbox.indices.IndicesExists;
 import io.searchbox.indices.mapping.GetMapping;
 import io.searchbox.indices.mapping.PutMapping;
-import org.apache.kafka.common.config.ConfigException;
-import org.apache.kafka.connect.data.Schema;
-import org.apache.kafka.connect.errors.ConnectException;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 import java.io.IOException;
 import java.util.ArrayList;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 public class JestElasticsearchClient implements ElasticsearchClient {
 
@@ -71,6 +71,8 @@ public class JestElasticsearchClient implements ElasticsearchClient {
   private final JestClient client;
   private final Version version;
 
+  private final Set<String> indexCache = new HashSet<>();
+
   // visible for testing
   public JestElasticsearchClient(JestClient client) {
     try {
@@ -187,6 +189,9 @@ public Version getVersion() {
   }
 
   private boolean indexExists(String index) {
+    if (indexCache.contains(index)) {
+      return true;
+    }
     Action action = new IndicesExists.Builder(index).build();
     try {
       JestResult result = client.execute(action);
@@ -209,6 +214,7 @@ public void createIndices(Set<String> indices) {
               throw new ConnectException("Could not create index '" + index + "'" + msg);
             }
           }
+          indexCache.add(index);
         } catch (IOException e) {
           throw new ConnectException(e);
         }

src/test/java/io/confluent/connect/elasticsearch/ElasticsearchSinkTaskTest.java

@@ -22,6 +22,12 @@
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.sink.SinkRecord;
+import org.elasticsearch.action.ActionFuture;
+import org.elasticsearch.action.ActionRequest;
+import org.elasticsearch.action.ActionRequestValidationException;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsRequest;
+import org.elasticsearch.action.admin.indices.exists.indices.IndicesExistsResponse;
+import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.InternalTestCluster;
 import org.junit.Test;
@@ -123,28 +129,39 @@ public void testCreateAndWriteToIndexNotCreatedAtStartTime() {
     cluster.ensureAtLeastNumDataNodes(3);
     Map<String, String> props = createProps();
 
+    props.put(ElasticsearchSinkConnectorConfig.AUTO_CREATE_INDICES_AT_START_CONFIG, "false");
+
     ElasticsearchSinkTask task = new ElasticsearchSinkTask();
 
     String key = "key";
     Schema schema = createSchema();
     Struct record = createRecord(schema);
 
     SinkRecord sinkRecord = new SinkRecord(UNSEEN_TOPIC,
-            PARTITION_114,
-            Schema.STRING_SCHEMA,
-            key,
-            schema,
-            record,
-            0 );
+        PARTITION_114,
+        Schema.STRING_SCHEMA,
+        key,
+        schema,
+        record,
+        0 );
 
-    try {
-      task.start(props, client);
-      task.open(new HashSet<>(Collections.singletonList(TOPIC_IN_CAPS_PARTITION)));
-      task.put(Collections.singleton(sinkRecord));
-    } catch (Exception ex) {
-      fail("Record could not be written to elasticsearch due to non existing index");
-    } finally {
-      task.stop();
-    }
+    task.start(props, client);
+    task.open(new HashSet<>(Collections.singletonList(TOPIC_IN_CAPS_PARTITION)));
+    task.put(Collections.singleton(sinkRecord));
+    task.stop();
+
+    assertTrue(UNSEEN_TOPIC + " index created without errors ",
+        verifyIndexExist(cluster, UNSEEN_TOPIC.toLowerCase()));
+
+  }
+
+  private boolean verifyIndexExist(InternalTestCluster cluster, String ... indices) {
+    ActionFuture<IndicesExistsResponse> action = cluster
+        .client()
+        .admin()
+        .indices()
+        .exists(new IndicesExistsRequest(indices));
+
+    return action.actionGet().isExists();
   }
 }