Add an option to disable index creation at bootstrap time, so index are created once every request to write an record is performed

Author: purbon

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

@@ -139,6 +139,11 @@ 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 "
+      + " of the Kafka topics.";
+
   protected static ConfigDef baseConfigDef() {
     final ConfigDef configDef = new ConfigDef();
     addConnectorConfigs(configDef);
@@ -244,10 +249,20 @@ private static void addConnectorConfigs(ConfigDef configDef) {
         3000, 
         Importance.LOW, 
         READ_TIMEOUT_MS_CONFIG_DOC,
-        group, 
-        ++order, 
-        Width.SHORT, 
-        "Read Timeout");
+        group,
+        ++order,
+        Width.SHORT,
+        "Read Timeout"
+    ).define(
+        CREATE_INDEX_AT_START_CONFIG,
+        Type.BOOLEAN,
+        true,
+        Importance.LOW,
+        CREATE_INDEX_AT_START__DOC,
+        group,
+        ++order,
+        Width.SHORT,
+        "Create indexes at bootstrap time");
   }
 
   private static void addConversionConfigs(ConfigDef configDef) {

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

@@ -40,6 +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;
 
   @Override
   public String version() {
@@ -90,6 +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);
 
       DataConverter.BehaviorOnNullValues behaviorOnNullValues =
           DataConverter.BehaviorOnNullValues.forValue(
@@ -136,6 +139,8 @@ public void start(Map<String, String> props, ElasticsearchClient client) {
           .setBehaviorOnNullValues(behaviorOnNullValues)
           .setBehaviorOnMalformedDoc(behaviorOnMalformedDoc);
 
+      this.createIndexAtStartTime = createIndexAtStartTime;
+
       writer = builder.build();
       writer.start();
     } catch (ConfigException e) {
@@ -149,11 +154,13 @@ 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);
-    Set<String> topics = new HashSet<>();
-    for (TopicPartition tp : partitions) {
-      topics.add(tp.topic());
+    if (createIndexAtStartTime) {
+      Set<String> topics = new HashSet<>();
+      for (TopicPartition tp : partitions) {
+        topics.add(tp.topic());
+      }
+      writer.createIndicesForTopics(topics);
     }
-    writer.createIndicesForTopics(topics);
   }
 
   @Override

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

@@ -253,6 +253,8 @@ public void write(Collection<SinkRecord> records) {
       final boolean ignoreSchema =
           ignoreSchemaTopics.contains(sinkRecord.topic()) || this.ignoreSchema;
 
+      client.createIndices(Collections.singleton(index));
+
       if (!ignoreSchema && !existingMappings.contains(index)) {
         try {
           if (Mapping.getMapping(client, index, type) == null) {

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

@@ -42,6 +42,11 @@ public class ElasticsearchSinkTaskTest extends ElasticsearchSinkTestBase {
   private static final int PARTITION_113 = 113;
   private static final TopicPartition TOPIC_IN_CAPS_PARTITION = new TopicPartition(TOPIC_IN_CAPS, PARTITION_113);
 
+  private static final String UNSEEN_TOPIC = "UnseenTopic";
+  private static final int PARTITION_114 = 114;
+  private static final TopicPartition UNSEEN_TOPIC_PARTITION = new TopicPartition(UNSEEN_TOPIC, PARTITION_114);
+
+
   private Map<String, String> createProps() {
     Map<String, String> props = new HashMap<>();
     props.put(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG, TYPE);
@@ -94,7 +99,38 @@ public void testCreateAndWriteToIndexForTopicWithUppercaseCharacters() {
     Struct record = createRecord(schema);
 
     SinkRecord sinkRecord = new SinkRecord(TOPIC_IN_CAPS,
-            PARTITION_113,
+        PARTITION_113,
+        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("A topic name not in lowercase can not be used as index name in Elasticsearch");
+    } finally {
+      task.stop();
+    }
+  }
+
+  @Test
+  public void testCreateAndWriteToIndexNotCreatedAtStartTime() {
+    InternalTestCluster cluster = ESIntegTestCase.internalCluster();
+    cluster.ensureAtLeastNumDataNodes(3);
+    Map<String, String> props = createProps();
+
+    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,
@@ -106,7 +142,7 @@ public void testCreateAndWriteToIndexForTopicWithUppercaseCharacters() {
       task.open(new HashSet<>(Collections.singletonList(TOPIC_IN_CAPS_PARTITION)));
       task.put(Collections.singleton(sinkRecord));
     } catch (Exception ex) {
-      fail("A topic name not in lowercase can not be used as index name in Elasticsearch");
+      fail("Record could not be written to elasticsearch due to non existing index");
     } finally {
       task.stop();
     }