Merge branch '3.2.x' into 3.3.x

Author: rhauch

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

@@ -225,8 +225,7 @@ public void write(Collection<SinkRecord> records) {
         continue;
       }
 
-      final String indexOverride = topicToIndexMap.get(sinkRecord.topic());
-      final String index = indexOverride != null ? indexOverride : sinkRecord.topic();
+      final String index = convertTopicToIndexName(sinkRecord.topic());
       final boolean ignoreKey = ignoreKeyTopics.contains(sinkRecord.topic()) || this.ignoreKey;
       final boolean ignoreSchema = ignoreSchemaTopics.contains(sinkRecord.topic()) || this.ignoreSchema;
 
@@ -261,6 +260,17 @@ public void write(Collection<SinkRecord> records) {
     }
   }
 
+  /**
+   * Return the expected index name for a given topic, using the configured mapping or the topic name. Elasticsearch
+   * <a href="https://github.com/elastic/elasticsearch/issues/29420">accepts only lowercase index names</a>.
+   */
+  private String convertTopicToIndexName(String topic) {
+    final String indexOverride = topicToIndexMap.get(topic);
+    String index = indexOverride != null ? indexOverride : topic.toLowerCase();
+    log.debug("Topic '{}' was translated as index '{}'", topic, index);
+    return index;
+  }
+
   public void flush() {
     bulkProcessor.flush(flushTimeoutMs);
   }
@@ -309,12 +319,7 @@ public void createIndicesForTopics(Set<String> assignedTopics) {
   private Set<String> indicesForTopics(Set<String> assignedTopics) {
     final Set<String> indices = new HashSet<>();
     for (String topic : assignedTopics) {
-      final String index = topicToIndexMap.get(topic);
-      if (index != null) {
-        indices.add(index);
-      } else {
-        indices.add(topic);
-      }
+      indices.add(convertTopicToIndexName(topic));
     }
     return indices;
   }

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

@@ -18,23 +18,30 @@
 
 import com.carrotsearch.randomizedtesting.annotations.ThreadLeakScope;
 
+import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.data.Struct;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.elasticsearch.test.ESIntegTestCase;
 import org.elasticsearch.test.InternalTestCluster;
 import org.junit.Test;
 
-import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
+import java.util.Collection;
+import java.util.ArrayList;
+import java.util.Collections;
+
 
 @ThreadLeakScope(ThreadLeakScope.Scope.NONE)
 public class ElasticsearchSinkTaskTest extends ElasticsearchSinkTestBase {
 
+  private static final String TOPIC_IN_CAPS = "AnotherTopicInCaps";
+  private static final int PARTITION_113 = 113;
+  private static final TopicPartition TOPIC_IN_CAPS_PARTITION = new TopicPartition(TOPIC_IN_CAPS, PARTITION_113);
+
   private Map<String, String> createProps() {
     Map<String, String> props = new HashMap<>();
     props.put(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG, TYPE);
@@ -72,4 +79,36 @@ public void testPutAndFlush() throws Exception {
     verifySearchResults(records, true, false);
   }
 
+  @Test
+  public void testCreateAndWriteToIndexForTopicWithUppercaseCharacters() {
+    // We should as well test that writing a record with a previously un seen record will create
+    // an index following the required elasticsearch requirements of lowercasing.
+    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(TOPIC_IN_CAPS,
+            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();
+    }
+  }
 }