Merge branch '5.0.x' into 5.1.x

Author: ConfluentJenkins

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

@@ -406,10 +406,6 @@ public static String[] names() {
       return result;
     }
 
-    public static BehaviorOnNullValues forValue(String value) {
-      return valueOf(value.toUpperCase(Locale.ROOT));
-    }
-
     @Override
     public String toString() {
       return name().toLowerCase(Locale.ROOT);

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

@@ -15,7 +15,6 @@
 
 package io.confluent.connect.elasticsearch;
 
-import io.confluent.connect.elasticsearch.bulk.BulkProcessor;
 import io.confluent.connect.elasticsearch.jest.JestElasticsearchClient;
 import org.apache.kafka.clients.consumer.OffsetAndMetadata;
 import org.apache.kafka.common.TopicPartition;
@@ -27,9 +26,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
-import java.util.HashMap;
 import java.util.HashSet;
-import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
@@ -58,61 +55,16 @@ public void start(Map<String, String> props, ElasticsearchClient client) {
       log.info("Starting ElasticsearchSinkTask.");
 
       ElasticsearchSinkConnectorConfig config = new ElasticsearchSinkConnectorConfig(props);
-      String type = config.getString(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG);
-      boolean ignoreKey =
-          config.getBoolean(ElasticsearchSinkConnectorConfig.KEY_IGNORE_CONFIG);
-      boolean ignoreSchema =
-          config.getBoolean(ElasticsearchSinkConnectorConfig.SCHEMA_IGNORE_CONFIG);
-      boolean useCompactMapEntries =
-          config.getBoolean(ElasticsearchSinkConnectorConfig.COMPACT_MAP_ENTRIES_CONFIG);
-
-
-      Map<String, String> topicToIndexMap =
-          parseMapConfig(config.getList(ElasticsearchSinkConnectorConfig.TOPIC_INDEX_MAP_CONFIG));
-      Set<String> topicIgnoreKey =
-          new HashSet<>(config.getList(ElasticsearchSinkConnectorConfig.TOPIC_KEY_IGNORE_CONFIG));
-      Set<String> topicIgnoreSchema = new HashSet<>(
-          config.getList(ElasticsearchSinkConnectorConfig.TOPIC_SCHEMA_IGNORE_CONFIG)
-      );
-
-      long flushTimeoutMs =
-          config.getLong(ElasticsearchSinkConnectorConfig.FLUSH_TIMEOUT_MS_CONFIG);
-      int maxBufferedRecords =
-          config.getInt(ElasticsearchSinkConnectorConfig.MAX_BUFFERED_RECORDS_CONFIG);
-      int batchSize =
-          config.getInt(ElasticsearchSinkConnectorConfig.BATCH_SIZE_CONFIG);
-      long lingerMs =
-          config.getLong(ElasticsearchSinkConnectorConfig.LINGER_MS_CONFIG);
-      int maxInFlightRequests =
-          config.getInt(ElasticsearchSinkConnectorConfig.MAX_IN_FLIGHT_REQUESTS_CONFIG);
-      long retryBackoffMs =
-          config.getLong(ElasticsearchSinkConnectorConfig.RETRY_BACKOFF_MS_CONFIG);
-      int maxRetry =
-          config.getInt(ElasticsearchSinkConnectorConfig.MAX_RETRIES_CONFIG);
-      boolean dropInvalidMessage =
-          config.getBoolean(ElasticsearchSinkConnectorConfig.DROP_INVALID_MESSAGE_CONFIG);
-      boolean createIndicesAtStartTime =
-          config.getBoolean(ElasticsearchSinkConnectorConfig.AUTO_CREATE_INDICES_AT_START_CONFIG);
-
-      DataConverter.BehaviorOnNullValues behaviorOnNullValues =
-          DataConverter.BehaviorOnNullValues.forValue(
-              config.getString(ElasticsearchSinkConnectorConfig.BEHAVIOR_ON_NULL_VALUES_CONFIG)
-          );
-
-      BulkProcessor.BehaviorOnMalformedDoc behaviorOnMalformedDoc =
-          BulkProcessor.BehaviorOnMalformedDoc.forValue(
-              config.getString(ElasticsearchSinkConnectorConfig.BEHAVIOR_ON_MALFORMED_DOCS_CONFIG)
-          );
 
       // Calculate the maximum possible backoff time ...
       long maxRetryBackoffMs =
-          RetryUtil.computeRetryWaitTimeInMillis(maxRetry, retryBackoffMs);
+          RetryUtil.computeRetryWaitTimeInMillis(config.maxRetries(), config.retryBackoffMs());
       if (maxRetryBackoffMs > RetryUtil.MAX_RETRY_TIME_MS) {
         log.warn("This connector uses exponential backoff with jitter for retries, "
                 + "and using '{}={}' and '{}={}' results in an impractical but possible maximum "
                 + "backoff time greater than {} hours.",
-            ElasticsearchSinkConnectorConfig.MAX_RETRIES_CONFIG, maxRetry,
-            ElasticsearchSinkConnectorConfig.RETRY_BACKOFF_MS_CONFIG, retryBackoffMs,
+            ElasticsearchSinkConnectorConfig.MAX_RETRIES_CONFIG, config.maxRetries(),
+            ElasticsearchSinkConnectorConfig.RETRY_BACKOFF_MS_CONFIG, config.retryBackoffMs(),
             TimeUnit.MILLISECONDS.toHours(maxRetryBackoffMs));
       }
 
@@ -123,23 +75,23 @@ public void start(Map<String, String> props, ElasticsearchClient client) {
       }
 
       ElasticsearchWriter.Builder builder = new ElasticsearchWriter.Builder(this.client)
-          .setType(type)
-          .setIgnoreKey(ignoreKey, topicIgnoreKey)
-          .setIgnoreSchema(ignoreSchema, topicIgnoreSchema)
-          .setCompactMapEntries(useCompactMapEntries)
-          .setTopicToIndexMap(topicToIndexMap)
-          .setFlushTimoutMs(flushTimeoutMs)
-          .setMaxBufferedRecords(maxBufferedRecords)
-          .setMaxInFlightRequests(maxInFlightRequests)
-          .setBatchSize(batchSize)
-          .setLingerMs(lingerMs)
-          .setRetryBackoffMs(retryBackoffMs)
-          .setMaxRetry(maxRetry)
-          .setDropInvalidMessage(dropInvalidMessage)
-          .setBehaviorOnNullValues(behaviorOnNullValues)
-          .setBehaviorOnMalformedDoc(behaviorOnMalformedDoc);
-
-      this.createIndicesAtStartTime = createIndicesAtStartTime;
+          .setType(config.type())
+          .setIgnoreKey(config.ignoreKey(), config.ignoreKeyTopics())
+          .setIgnoreSchema(config.ignoreSchema(), config.ignoreSchemaTopics())
+          .setCompactMapEntries(config.useCompactMapEntries())
+          .setTopicToIndexMap(config.topicToIndexMap())
+          .setFlushTimoutMs(config.flushTimeoutMs())
+          .setMaxBufferedRecords(config.maxBufferedRecords())
+          .setMaxInFlightRequests(config.maxInFlightRequests())
+          .setBatchSize(config.batchSize())
+          .setLingerMs(config.lingerMs())
+          .setRetryBackoffMs(config.retryBackoffMs())
+          .setMaxRetry(config.maxRetries())
+          .setDropInvalidMessage(config.dropInvalidMessage())
+          .setBehaviorOnNullValues(config.behaviorOnNullValues())
+          .setBehaviorOnMalformedDoc(config.behaviorOnMalformedDoc());
+
+      this.createIndicesAtStartTime = config.createIndicesAtStart();
 
       writer = builder.build();
       writer.start();
@@ -190,16 +142,4 @@ public void stop() throws ConnectException {
       client.close();
     }
   }
-
-  private Map<String, String> parseMapConfig(List<String> values) {
-    Map<String, String> map = new HashMap<>();
-    for (String value : values) {
-      String[] parts = value.split(":");
-      String topic = parts[0];
-      String type = parts[1];
-      map.put(topic, type);
-    }
-    return map;
-  }
-
 }

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

@@ -532,10 +532,6 @@ public static String[] names() {
       return result;
     }
 
-    public static BehaviorOnMalformedDoc forValue(String value) {
-      return valueOf(value.toUpperCase(Locale.ROOT));
-    }
-
     @Override
     public String toString() {
       return name().toLowerCase(Locale.ROOT);

src/main/java/io/confluent/connect/elasticsearch/bulk/BulkProcessor.java

@@ -49,7 +49,6 @@
 import io.searchbox.indices.mapping.PutMapping;
 import org.apache.http.HttpHost;
 import org.apache.kafka.common.config.ConfigException;
-import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.slf4j.Logger;
@@ -126,26 +125,18 @@ public JestElasticsearchClient(Map<String, String> props) {
   protected JestElasticsearchClient(Map<String, String> props, JestClientFactory factory) {
     try {
       ElasticsearchSinkConnectorConfig config = new ElasticsearchSinkConnectorConfig(props);
-      final int connTimeout = config.getInt(
-          ElasticsearchSinkConnectorConfig.CONNECTION_TIMEOUT_MS_CONFIG);
-      final int readTimeout = config.getInt(
-          ElasticsearchSinkConnectorConfig.READ_TIMEOUT_MS_CONFIG);
-
-      final String username = config.getString(
-          ElasticsearchSinkConnectorConfig.CONNECTION_USERNAME_CONFIG);
-      final Password password = config.getPassword(
-          ElasticsearchSinkConnectorConfig.CONNECTION_PASSWORD_CONFIG);
-
-      List<String> address =
-          config.getList(ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG);
-      HttpClientConfig.Builder builder = new HttpClientConfig.Builder(address)
-          .connTimeout(connTimeout)
-          .readTimeout(readTimeout)
+
+      Set<String> addresses = config.connectionUrls();
+      HttpClientConfig.Builder builder = new HttpClientConfig.Builder(addresses)
+          .connTimeout(config.connectionTimeoutMs())
+          .readTimeout(config.readTimeoutMs())
           .multiThreaded(true);
-      if (username != null && password != null) {
-        builder.defaultCredentials(username, password.value())
-            .preemptiveAuthTargetHosts(address.stream()
-                .map(addr -> HttpHost.create(addr)).collect(Collectors.toSet()));
+      if (config.isAuthenticatedConnection()) {
+        builder
+            .defaultCredentials(config.username(), config.password().value())
+            .preemptiveAuthTargetHosts(
+                addresses.stream().map(addr -> HttpHost.create(addr)).collect(Collectors.toSet())
+          );
       }
       HttpClientConfig httpClientConfig = builder.build();
       factory.setHttpClientConfig(httpClientConfig);

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

@@ -1,6 +1,7 @@
 package io.confluent.connect.elasticsearch;
 
-import org.junit.Assert;
+import static org.junit.Assert.assertEquals;
+
 import org.junit.Before;
 import org.junit.Test;
 
@@ -16,34 +17,22 @@ public void setup() {
     props = new HashMap<>();
     props.put(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG, ElasticsearchSinkTestBase.TYPE);
     props.put(ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG, "localhost");
-    props.put(ElasticsearchSinkConnectorConfig.KEY_IGNORE_CONFIG, "true");
+    props.put(ElasticsearchSinkConnectorConfig.IGNORE_KEY_CONFIG, "true");
   }
 
   @Test
   public void testDefaultHttpTimeoutsConfig() {
     ElasticsearchSinkConnectorConfig config = new ElasticsearchSinkConnectorConfig(props);
-    Assert.assertEquals(
-        config.getInt(ElasticsearchSinkConnectorConfig.READ_TIMEOUT_MS_CONFIG),
-        (Integer) 3000
-    );
-    Assert.assertEquals(
-        config.getInt(ElasticsearchSinkConnectorConfig.CONNECTION_TIMEOUT_MS_CONFIG),
-        (Integer) 1000
-    );
+    assertEquals(config.readTimeoutMs(), 3000);
+    assertEquals(config.connectionTimeoutMs(), 1000);
   }
 
   @Test
   public void testSetHttpTimeoutsConfig() {
     props.put(ElasticsearchSinkConnectorConfig.READ_TIMEOUT_MS_CONFIG, "10000");
     props.put(ElasticsearchSinkConnectorConfig.CONNECTION_TIMEOUT_MS_CONFIG, "15000");
     ElasticsearchSinkConnectorConfig config = new ElasticsearchSinkConnectorConfig(props);
-    Assert.assertEquals(
-        config.getInt(ElasticsearchSinkConnectorConfig.READ_TIMEOUT_MS_CONFIG),
-        (Integer) 10000
-    );
-    Assert.assertEquals(
-        config.getInt(ElasticsearchSinkConnectorConfig.CONNECTION_TIMEOUT_MS_CONFIG),
-        (Integer) 15000
-    );
+    assertEquals(config.readTimeoutMs(), 10000);
+    assertEquals(config.connectionTimeoutMs(), 15000);
   }
 }

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

@@ -22,11 +22,8 @@
 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;
@@ -56,7 +53,7 @@ private Map<String, String> createProps() {
     Map<String, String> props = new HashMap<>();
     props.put(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG, TYPE);
     props.put(ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG, "localhost");
-    props.put(ElasticsearchSinkConnectorConfig.KEY_IGNORE_CONFIG, "true");
+    props.put(ElasticsearchSinkConnectorConfig.IGNORE_KEY_CONFIG, "true");
     return props;
   }
 
@@ -128,7 +125,7 @@ public void testCreateAndWriteToIndexNotCreatedAtStartTime() {
     cluster.ensureAtLeastNumDataNodes(3);
     Map<String, String> props = createProps();
 
-    props.put(ElasticsearchSinkConnectorConfig.AUTO_CREATE_INDICES_AT_START_CONFIG, "false");
+    props.put(ElasticsearchSinkConnectorConfig.CREATE_INDICES_AT_START_CONFIG, "false");
 
     ElasticsearchSinkTask task = new ElasticsearchSinkTask();
 

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

@@ -17,16 +17,12 @@
 
 import static org.junit.Assert.assertNotNull;
 
-import com.google.gson.JsonObject;
 import io.confluent.common.utils.IntegrationTest;
 import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig;
 import io.confluent.connect.elasticsearch.ElasticsearchSinkTask;
-import io.confluent.connect.elasticsearch.Mapping;
-import io.confluent.connect.elasticsearch.TestUtils;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
@@ -43,7 +39,7 @@ private Map<String, String> createProps() {
     Map<String, String> props = new HashMap<>();
     props.put(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG, TYPE);
     props.put(ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG, "localhost");
-    props.put(ElasticsearchSinkConnectorConfig.KEY_IGNORE_CONFIG, "true");
+    props.put(ElasticsearchSinkConnectorConfig.IGNORE_KEY_CONFIG, "true");
     return props;
   }