Merge pull request #880 from confluentinc/backport-pr870-to-14.1.x

CC-32762: Set threadName with taskId prefix

Author: jjain1259

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

@@ -32,6 +32,7 @@
 import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
+import java.util.concurrent.ThreadFactory;
 import java.util.function.BiConsumer;
 
 import org.apache.http.HttpHost;
@@ -122,9 +123,21 @@ public class ElasticsearchClient {
   public ElasticsearchClient(
       ElasticsearchSinkConnectorConfig config,
       ErrantRecordReporter reporter,
-      Runnable afterBulkCallback
+      Runnable afterBulkCallback,
+      int taskId,
+      String connectorName
   ) {
-    this.bulkExecutorService = Executors.newFixedThreadPool(config.maxInFlightRequests());
+    this.bulkExecutorService = Executors.newFixedThreadPool(config.maxInFlightRequests(),
+      new ThreadFactory() {
+        private final AtomicInteger threadNumber = new AtomicInteger(1);
+        @Override
+        public Thread newThread(Runnable r) {
+          Thread thread = Executors.defaultThreadFactory().newThread(r);
+          thread.setName(connectorName + "-" + taskId + "-elasticsearch-bulk-executor-"
+                  + threadNumber.getAndIncrement());
+          return thread;
+        }
+      });
     this.numBufferedRecords = new AtomicInteger(0);
     this.error = new AtomicReference<>();
     this.requestToSinkRecord = new ConcurrentHashMap<>();

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

@@ -58,10 +58,10 @@ public Class<? extends Task> taskClass() {
   @Override
   public List<Map<String, String>> taskConfigs(int maxTasks) {
     List<Map<String, String>> taskConfigs = new ArrayList<>();
-    Map<String, String> taskProps = new HashMap<>();
-    taskProps.putAll(configProperties);
     for (int i = 0; i < maxTasks; i++) {
-      taskConfigs.add(taskProps);
+      HashMap<String, String> taskConfig = new HashMap<>(configProperties);
+      taskConfig.put(ElasticsearchSinkTaskConfig.TASK_ID_CONFIG, Integer.toString(i));
+      taskConfigs.add(taskConfig);
     }
     return taskConfigs;
   }

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

@@ -891,6 +891,10 @@ private static void addDataStreamConfigs(ConfigDef configDef) {
 
   public static final ConfigDef CONFIG = baseConfigDef();
 
+  protected ElasticsearchSinkConnectorConfig(ConfigDef config, Map<String, String> properties) {
+    super(config, properties);
+  }
+
   public ElasticsearchSinkConnectorConfig(Map<String, String> props) {
     super(CONFIG, props);
   }

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

@@ -42,7 +42,7 @@ public class ElasticsearchSinkTask extends SinkTask {
 
   private DataConverter converter;
   private ElasticsearchClient client;
-  private ElasticsearchSinkConnectorConfig config;
+  private ElasticsearchSinkTaskConfig config;
   private ErrantRecordReporter reporter;
   private Set<String> existingMappings;
   private Set<String> indexCache;
@@ -58,7 +58,7 @@ public void start(Map<String, String> props) {
   protected void start(Map<String, String> props, ElasticsearchClient client) {
     log.info("Starting ElasticsearchSinkTask.");
 
-    this.config = new ElasticsearchSinkConnectorConfig(props);
+    this.config = new ElasticsearchSinkTaskConfig(props);
     this.converter = new DataConverter(config);
     this.existingMappings = new HashSet<>();
     this.indexCache = new HashSet<>();
@@ -80,7 +80,8 @@ protected void start(Map<String, String> props, ElasticsearchClient client) {
     }
     Runnable afterBulkCallback = () -> offsetTracker.updateOffsets();
     this.client = client != null ? client
-        : new ElasticsearchClient(config, reporter, afterBulkCallback);
+        : new ElasticsearchClient(config, reporter, afterBulkCallback,
+            config.getTaskId(), config.getConnectorName());
 
     if (!config.flushSynchronously()) {
       this.offsetTracker = new AsyncOffsetTracker(context);

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

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2018 Confluent Inc.
+ *
+ * Licensed under the Confluent Community License (the "License"); you may not use
+ * this file except in compliance with the License.  You may obtain a copy of the
+ * License at
+ *
+ * http://www.confluent.io/confluent-community-license
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations under the License.
+ */
+
+package io.confluent.connect.elasticsearch;
+
+import org.apache.kafka.common.config.ConfigDef;
+import java.util.Map;
+
+public class ElasticsearchSinkTaskConfig extends ElasticsearchSinkConnectorConfig {
+
+  private final int taskId;
+  private final String connectorName;
+
+  public static final String TASK_ID_CONFIG =                   "taskId";
+  private static final ConfigDef.Type TASK_ID_TYPE =            ConfigDef.Type.INT;
+  public static final ConfigDef.Importance TASK_ID_IMPORTANCE = ConfigDef.Importance.LOW;
+  public static final int TASK_ID_DEFAULT = 0;
+
+  /**
+   * Return a ConfigDef object used to define this config's fields.
+   *
+   * @return A ConfigDef object used to define this config's fields.
+   */
+  public static ConfigDef config() {
+    return ElasticsearchSinkConnectorConfig.baseConfigDef()
+            .defineInternal(
+                    TASK_ID_CONFIG,
+                    TASK_ID_TYPE,
+                    TASK_ID_DEFAULT,
+                    TASK_ID_IMPORTANCE
+            );
+  }
+
+  /**
+   * @param properties A Map detailing configuration properties and their respective values.
+   */
+  public ElasticsearchSinkTaskConfig(Map<String, String> properties) {
+    super(config(), properties);
+    taskId = properties.containsKey(TASK_ID_CONFIG) ? getInt(TASK_ID_CONFIG) : 0;
+    connectorName = originalsStrings().get("name");
+  }
+
+  public int getTaskId() {
+    return taskId;
+  }
+
+  public String getConnectorName() {
+    return connectorName;
+  }
+}