Unverified Commit e52099de authored by Randall Hauch's avatar Randall Hauch Committed by GitHub
Browse files

Merge pull request #251 from purbon/fix/elastic-lowercase-indexname 

[Backport ready] Fix bug when topic names are not in all lowercase
parents 26984c3c 72f996ec

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

+13 −8
Original line number Diff line number Diff line
@@ -190,8 +190,7 @@ public class ElasticsearchWriter { 


  public void write(Collection<SinkRecord> records) {

    for (SinkRecord sinkRecord : records) {

      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;
@@ -213,6 +212,17 @@ public class ElasticsearchWriter { 
    }

  }



  /**

   * 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);

  }
@@ -261,12 +271,7 @@ public class ElasticsearchWriter { 
  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

+41 −2
Original line number Diff line number Diff line
@@ -18,6 +18,7 @@ package io.confluent.connect.elasticsearch; 


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;
@@ -25,16 +26,22 @@ 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 class ElasticsearchSinkTaskTest extends ElasticsearchSinkTestBase { 
    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();

    }

  }

}