Treating threads as daemons (#10)

* treating threads as daemons, set threads names * set thread number * setup gradle release * adjusting id * consumer stop can't hang application * [Gradle Release Plugin] - new version commit: '2.0.6-snapshot'. * version adjusting

Author: mageddo

README.md

@@ -20,7 +20,7 @@ Kafka Client is a vanilla java library that makes it easy to consume data from k
 # Getting Started
 
 ```groovy
-compile("com.mageddo.rapids-kafka-client:rapids-kafka-client:2.0.4")
+compile("com.mageddo.rapids-kafka-client:rapids-kafka-client:2.0.5-snapshot")
 ```
 
 ```java

build.gradle

@@ -1,6 +1,7 @@
 plugins {
  id "java-library"
  id "maven"
+ id 'net.researchgate.release' version '2.8.1'
 }
 
 apply from: "nexus.gradle"
@@ -37,3 +38,19 @@ compileTestJava {
  sourceCompatibility = JavaVersion.VERSION_11
  targetCompatibility = JavaVersion.VERSION_11
 }
+
+release {
+ project.ext.set("release.useAutomaticVersion", true)
+ git {
+ requireBranch = ''
+ }
+ failOnCommitNeeded = false
+ failOnPublishNeeded = false
+ failOnUnversionedFiles = false
+ buildTasks = []
+}
+
+confirmReleaseVersion.doLast {
+ def f = file("${project.projectDir}/README.md")
+ f.text = f.text.replaceAll("(:\\ )*\\d+\\.\\d+\\.\\d+", "\$1${version}")
+}

gradle.properties

@@ -1,4 +1,4 @@
 projectUrl=https://github.com/mageddo-projects/kafka-client
 group=com.mageddo.rapids-kafka-client
-version=2.0.4
+version=2.0.6
 org.gradle.daemon=false

src/main/java/com/mageddo/kafka/client/BatchConsumer.java

@@ -20,6 +20,7 @@ public class BatchConsumer<K, V> extends DefaultConsumer<K, V> {
 
  private final Consumer<K, V> consumer;
  private final ConsumerConfig<K, V> consumerConfig;
+ private final int number;
 
  @Override
  protected void consume(ConsumerRecords<K, V> records) {
@@ -78,6 +79,11 @@ protected ConsumerConfig<K, V> consumerConfig() {
  return this.consumerConfig;
  }
 
+ @Override
+ protected int getNumber() {
+ return this.number;
+ }
+
  private void commitFirstRecord(Consumer<K, V> consumer, ConsumerRecords<K, V> records, TopicPartition partition) {
  final ConsumerRecord<K, V> firstRecord = getFirstRecord(records, partition);
  if (firstRecord != null) {

src/main/java/com/mageddo/kafka/client/ConsumerController.java

@@ -75,7 +75,7 @@ public void consume(ConsumerConfig<K, V> consumerConfig) {
  this.checkReasonablePollInterval(consumerConfig);
 
  for (int i = 0; i < threads; i++) {
- final ThreadConsumer<K, V> consumer = this.getInstance(this.create(consumerConfig), consumerConfig);
+ final ThreadConsumer<K, V> consumer = this.getInstance(this.create(consumerConfig), consumerConfig, i);
  consumer.start();
  }
  log.info(
@@ -84,11 +84,11 @@ public void consume(ConsumerConfig<K, V> consumerConfig) {
  );
  }
 
- ThreadConsumer<K, V> getInstance(Consumer<K, V> consumer, ConsumerConfig<K, V> consumerConfig) {
+ ThreadConsumer<K, V> getInstance(Consumer<K, V> consumer, ConsumerConfig<K, V> consumerConfig, int number) {
  if (consumerConfig.batchCallback() != null) {
- return this.bindInstance(new BatchConsumer<>(consumer, consumerConfig));
+ return this.bindInstance(new BatchConsumer<>(consumer, consumerConfig, number));
  }
- return this.bindInstance(new RecordConsumer<>(consumer, consumerConfig));
+ return this.bindInstance(new RecordConsumer<>(consumer, consumerConfig, number));
  }
 
  private ThreadConsumer<K, V> bindInstance(ThreadConsumer<K, V> consumer) {

src/main/java/com/mageddo/kafka/client/ConsumerStarter.java

@@ -2,12 +2,12 @@
 
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
+import com.mageddo.kafka.client.internal.Threads;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -102,33 +102,23 @@ public void stop() {
  }
  this.stopped = true;
  log.info("status=stopping-consumers, toStop={}", this.factories.size());
- final ExecutorService executorService = Executors.newScheduledThreadPool(5);
- try {
- final List<Future<String>> futures = new ArrayList<>();
- for (ConsumerController<?, ?> factory : this.factories) {
- futures.add(executorService.submit(() -> {
- try {
- factory.close();
- return factory.toString();
- } catch (Exception e) {
- log.warn("status=failed-to-stop-consumer, consumer={}", factory);
- return String.format("failed to stop: %s, %s", factory.toString(), e.getMessage());
- }
- }));
- }
- int stopped = 0;
- for (Future<String> future : futures) {
+ final ExecutorService executorService = Threads.createPool(5);
+ final AtomicInteger stopped = new AtomicInteger(1);
+ for (int i = 0; i < this.factories.size(); i++) {
+ final ConsumerController<?, ?> factory = this.factories.get(i);
+ executorService.submit(() -> {
  try {
- final String id = future.get();
- log.info("status=stopped, {} of {}, factory={}", ++stopped, this.factories.size(), id);
- } catch (InterruptedException | ExecutionException e) {
- throw new RuntimeException(e);
+ factory.close();
+ log.info(
+ "status=stopped, {} of {}, factory={}",
+ stopped.getAndIncrement(), this.factories.size(), factory
+ );
+ return factory.toString();
+ } catch (Exception e) {
+ log.warn("status=failed-to-stop-consumer, consumer={}", factory);
+ return String.format("failed to stop: %s, %s", factory.toString(), e.getMessage());
  }
- }
- } finally {
- executorService.shutdown();
- log.info("status=consumers-stopped, count={}", this.factories.size());
-
+ });
  }
  }
 

src/main/java/com/mageddo/kafka/client/DefaultConsumer.java

@@ -6,6 +6,7 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import com.mageddo.kafka.client.internal.ObjectsUtils;
+import com.mageddo.kafka.client.internal.Threads;
 
 import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
@@ -18,6 +19,7 @@
 import lombok.extern.slf4j.Slf4j;
 
 import static com.mageddo.kafka.client.DefaultConsumingConfig.DEFAULT_POLL_TIMEOUT;
+import static com.mageddo.kafka.client.internal.StringUtils.clearNonAlpha;
 
 @Slf4j
 public abstract class DefaultConsumer<K, V> implements ThreadConsumer {
@@ -34,14 +36,17 @@ public abstract class DefaultConsumer<K, V> implements ThreadConsumer {
 
  protected abstract ConsumerConfig<K, V> consumerConfig();
 
+ protected abstract int getNumber();
+
  @Override
  public void start() {
  if (started.get()) {
  log.warn("status=already-started, thread={}, config={}", this.id(), this.consumerConfig());
  return;
  }
  final Consumer<K, V> consumer = consumer();
- this.executor = new Thread(() -> {
+
+ this.executor = newThread(() -> {
  this.poll(consumer, consumerConfig());
  });
  this.executor.start();
@@ -100,7 +105,7 @@ public void close() {
 
  @Override
  public String id() {
- return String.format("%d-%s_%s", this.executor.getId(), this.executor.getName(), this.consumerConfig());
+ return String.format("%d-%s", this.executor.getId(), this.executor.getName());
  }
 
  public boolean isClosed() {
@@ -158,6 +163,19 @@ void doRecoverWhenAvailable(RecoverContext<K, V> ctx, RecoverCallback<K, V> reco
  }
 
  Exception getConsumerError() {
- return consumerError;
+ return this.consumerError;
+ }
+
+ private Thread newThread(Runnable r) {
+ return Threads.newThread(r, this.createThreadId());
  }
+
+ private String createThreadId() {
+ return String.format(
+ "%s-%d",
+ clearNonAlpha(consumerConfig().groupId()),
+ this.getNumber()
+ );
+ }
+
 }

src/main/java/com/mageddo/kafka/client/RecordConsumer.java

@@ -19,6 +19,7 @@ public class RecordConsumer<K, V> extends DefaultConsumer<K, V> {
 
  private final Consumer<K, V> consumer;
  private final ConsumerConfig<K, V> consumerConfig;
+ private final int number;
 
  @Override
  protected void consume(ConsumerRecords<K, V> records) {
@@ -89,6 +90,11 @@ protected ConsumerConfig<K, V> consumerConfig() {
  return this.consumerConfig;
  }
 
+ @Override
+ protected int getNumber() {
+ return this.number;
+ }
+
  RetryPolicy getRetryPolicy() {
  return ObjectsUtils.firstNonNull(this.consumerConfig.retryPolicy(), DEFAULT_RETRY_STRATEGY);
  }

src/main/java/com/mageddo/kafka/client/internal/StringUtils.java

@@ -0,0 +1,14 @@
+package com.mageddo.kafka.client.internal;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class StringUtils {
+ public static String clearNonAlpha(String text) {
+ if (text == null) {
+ return null;
+ }
+ return text.replaceAll("^[a-zA-Z_\\-.]", "");
+ }
+}

src/main/java/com/mageddo/kafka/client/internal/Threads.java

@@ -0,0 +1,32 @@
+package com.mageddo.kafka.client.internal;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import lombok.AccessLevel;
+import lombok.NoArgsConstructor;
+
+@NoArgsConstructor(access = AccessLevel.PRIVATE)
+public class Threads {
+ public static ExecutorService createPool(int size) {
+ return Executors.newFixedThreadPool(size, r -> {
+ Thread t = Executors.defaultThreadFactory()
+ .newThread(r);
+ t.setDaemon(true);
+ return t;
+ });
+ }
+
+ public static Thread newThread(Runnable r) {
+ return newThread(r, null);
+ }
+
+ public static Thread newThread(Runnable r, String id) {
+ final Thread t = new Thread(r);
+ if (id != null) {
+ t.setName(id);
+ }
+ t.setDaemon(true);
+ return t;
+ }
+}