Merge branch '5.3.x' into 5.4.x

Lev Zemlyanov · Lev Zemlyanov · commit 28feb9d051eb · 2020-09-29T16:26:24.000-07:00
diff --git a/src/main/java/io/confluent/connect/elasticsearch/DataConverter.java b/src/main/java/io/confluent/connect/elasticsearch/DataConverter.java
@@ -428,10 +428,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);
diff --git a/src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfig.java b/src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfig.java
diff --git a/src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkTask.java b/src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkTask.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,29 +75,29 @@ 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();
       log.info(
           "Started ElasticsearchSinkTask, will {} records with null values ('{}')",
-          behaviorOnNullValues,
+          config.behaviorOnNullValues().name(),
           ElasticsearchSinkConnectorConfig.BEHAVIOR_ON_NULL_VALUES_CONFIG
       );
     } catch (ConfigException e) {
@@ -195,16 +147,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;
-  }
-
 }
diff --git a/src/main/java/io/confluent/connect/elasticsearch/bulk/BulkProcessor.java b/src/main/java/io/confluent/connect/elasticsearch/bulk/BulkProcessor.java
@@ -584,10 +584,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);
diff --git a/src/main/java/io/confluent/connect/elasticsearch/jest/JestElasticsearchClient.java b/src/main/java/io/confluent/connect/elasticsearch/jest/JestElasticsearchClient.java
@@ -56,7 +56,6 @@
 import org.apache.http.conn.ssl.SSLConnectionSocketFactory;
 import org.apache.http.nio.conn.ssl.SSLIOSessionStrategy;
 import org.apache.kafka.common.config.ConfigException;
-import org.apache.kafka.common.config.types.Password;
 import org.apache.kafka.common.network.Mode;
 import org.apache.kafka.common.security.ssl.SslFactory;
 import org.apache.kafka.common.utils.SystemTime;
@@ -149,12 +148,10 @@ protected JestElasticsearchClient(Map<String, String> props, JestClientFactory f
       factory.setHttpClientConfig(getClientConfig(config));
       this.client = factory.getObject();
       this.version = getServerVersion();
-      this.writeMethod = WriteMethod.forValue(
-              config.getString(ElasticsearchSinkConnectorConfig.WRITE_METHOD_CONFIG));
-      this.retryBackoffMs =
-              config.getLong(ElasticsearchSinkConnectorConfig.RETRY_BACKOFF_MS_CONFIG);
-      this.maxRetries = config.getInt(ElasticsearchSinkConnectorConfig.MAX_RETRIES_CONFIG);
-      this.timeout = config.getInt(ElasticsearchSinkConnectorConfig.READ_TIMEOUT_MS_CONFIG);
+      this.writeMethod = config.writeMethod();
+      this.retryBackoffMs = config.retryBackoffMs();
+      this.maxRetries = config.maxRetries();
+      this.timeout = config.readTimeoutMs();
     } catch (IOException e) {
       throw new ConnectException(
           "Couldn't start ElasticsearchSinkTask due to connection error:",
@@ -170,44 +167,35 @@ protected JestElasticsearchClient(Map<String, String> props, JestClientFactory f
 
   // Visible for Testing
   public static HttpClientConfig getClientConfig(ElasticsearchSinkConnectorConfig config) {
-    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);
-
-    final int maxInFlightRequests = config.getInt(
-        ElasticsearchSinkConnectorConfig.MAX_IN_FLIGHT_REQUESTS_CONFIG);
 
+    Set<String> addresses = config.connectionUrls();
     HttpClientConfig.Builder builder =
-        new HttpClientConfig.Builder(address)
-            .connTimeout(connTimeout)
-            .readTimeout(readTimeout)
-            .defaultMaxTotalConnectionPerRoute(maxInFlightRequests)
+        new HttpClientConfig.Builder(addresses)
+            .connTimeout(config.connectionTimeoutMs())
+            .readTimeout(config.readTimeoutMs())
+            .defaultMaxTotalConnectionPerRoute(config.maxInFlightRequests())
             .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())
+        );
     }
 
     if (config.secured()) {
-      log.info("Using secured connection to {}", address);
+      log.info("Using secured connection to {}", addresses);
       configureSslContext(builder, config);
     } else {
-      log.info("Using unsecured connection to {}", address);
+      log.info("Using unsecured connection to {}", addresses);
     }
     return builder.build();
   }
 
-  private static void configureSslContext(HttpClientConfig.Builder builder,
-                                            ElasticsearchSinkConnectorConfig config) {
+  private static void configureSslContext(
+      HttpClientConfig.Builder builder,
+      ElasticsearchSinkConnectorConfig config
+  ) {
+
     SslFactory kafkaSslFactory = new SslFactory(Mode.CLIENT, null, false);
     kafkaSslFactory.configure(config.sslConfigs());
 
@@ -649,10 +637,6 @@ public static String[] names() {
       return new String[] {INSERT.toString(), UPSERT.toString()};
     }
 
-    public static WriteMethod forValue(String value) {
-      return valueOf(value.toUpperCase(Locale.ROOT));
-    }
-
     @Override
     public String toString() {
       return name().toLowerCase(Locale.ROOT);
diff --git a/src/test/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfigTest.java b/src/test/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfigTest.java
@@ -1,7 +1,8 @@
 package io.confluent.connect.elasticsearch;
 
+import static org.junit.Assert.assertEquals;
+
 import org.apache.kafka.common.config.types.Password;
-import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -20,35 +21,23 @@ 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);
   }
 
   @Test
@@ -59,17 +48,17 @@ public void testSslConfigs() {
     props.put("elastic.https.ssl.truststore.password", "opensesame2");
     ElasticsearchSinkConnectorConfig config = new ElasticsearchSinkConnectorConfig(props);
     Map<String, Object> sslConfigs = config.sslConfigs();
-    Assert.assertTrue(sslConfigs.size() > 0);
-    Assert.assertEquals(
+    assertTrue(sslConfigs.size() > 0);
+    assertEquals(
         new Password("opensesame"),
         sslConfigs.get("ssl.keystore.password")
     );
-    Assert.assertEquals(
+    assertEquals(
         new Password("opensesame2"),
         sslConfigs.get("ssl.truststore.password")
     );
-    Assert.assertEquals("/path", sslConfigs.get("ssl.keystore.location"));
-    Assert.assertEquals("/path2", sslConfigs.get("ssl.truststore.location"));
+    assertEquals("/path", sslConfigs.get("ssl.keystore.location"));
+    assertEquals("/path2", sslConfigs.get("ssl.truststore.location"));
   }
 
   @Test
diff --git a/src/test/java/io/confluent/connect/elasticsearch/ElasticsearchSinkTaskTest.java b/src/test/java/io/confluent/connect/elasticsearch/ElasticsearchSinkTaskTest.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");
     props.put(ElasticsearchSinkConnectorConfig.READ_TIMEOUT_MS_CONFIG, "3000");
     return props;
   }
@@ -129,7 +126,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();
 
diff --git a/src/test/java/io/confluent/connect/elasticsearch/integration/ElasticsearchSinkTaskIT.java b/src/test/java/io/confluent/connect/elasticsearch/integration/ElasticsearchSinkTaskIT.java
@@ -57,7 +57,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, container.getConnectionUrl());
-    props.put(ElasticsearchSinkConnectorConfig.KEY_IGNORE_CONFIG, "true");
+    props.put(ElasticsearchSinkConnectorConfig.IGNORE_KEY_CONFIG, "true");
     return props;
   }
 

Original file line number	Diff line number	Diff line change
`@@ -57,7 +57,7 @@ private Map<String, String> createProps() {`
`57`	`57`	`Map<String, String> props = new HashMap<>();`
`58`	`58`	`props.put(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG, TYPE);`
`59`	`59`	`props.put(ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG, container.getConnectionUrl());`
`60`		`- props.put(ElasticsearchSinkConnectorConfig.KEY_IGNORE_CONFIG, "true");`
	`60`	`+ props.put(ElasticsearchSinkConnectorConfig.IGNORE_KEY_CONFIG, "true");`
`61`	`61`	`return props;`
`62`	`62`	`}`
`63`	`63`