Builder and default threads fix (#7)

* fixing generics * honoring disabled consumer * respect default consumer * fixing version

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.1")
+compile("com.mageddo.rapids-kafka-client:rapids-kafka-client:2.0.2")
 ```
 
 ```java

gradle.properties

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

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

@@ -1,7 +1,5 @@
 package com.mageddo.kafka.client;
 
-import com.mageddo.kafka.client.ConsumerConfig;
-
 public interface Consumer {
- ConsumerConfig<?, ?> config();
+ <K, V> ConsumerConfig<K, V> config();
 }

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

@@ -192,31 +192,48 @@ public static void waitFor() {
  .join();
  }
 
- public static Builder<?, ?> builderOf(ConsumerConfig<?, ?> config) {
- return copyToBuilder(config, ConsumerConfigDefault.builder().build());
+ public static <K, V> Builder<K, V> builderOf(ConsumerConfig<K, V> config) {
+ return copyToBuilder(config, ConsumerConfigDefault.<K, V>builder().build());
  }
 
- public static ConsumerConfigDefault<?, ?> copy(ConsumerConfig<?, ?> primary, ConsumerConfig<?, ?> secondary){
+ public static <K, V> ConsumerConfigDefault<K, V> copy(ConsumerConfig<K, V> primary, ConsumerConfig<K, V> secondary) {
  return copyToBuilder(primary, secondary).build();
  }
 
- public static Builder<?, ?> copyToBuilder(ConsumerConfig<?, ?> primary, ConsumerConfig<?, ?> secondary){
- final Builder builder = ConsumerConfigDefault.builder();
- fillWith(secondary, builder);
- fillWith(primary, builder);
+ public static <K, V> Builder<K, V> copyToBuilder(ConsumerConfig<K, V> primary, ConsumerConfig<K, V> secondary) {
+ final Builder<K, V> builder = copyToBuilder(
+ secondary,
+ ConsumerConfig.<K, V>builder().build(),
+ ConsumerConfig.builder()
+ );
+ return copyToBuilder(primary, secondary, builder);
+ }
+
+ public static <K, V> Builder<K, V> copyToBuilder(
+ ConsumerConfig<K, V> primary, ConsumerConfig<K, V> secondary, Builder<K, V> builder
+ ) {
+ fillPropsWith(secondary, builder);
+ fillPropsWith(primary, builder);
  return builder
  .callback(firstNonNull(primary.callback(), secondary.callback()))
  .batchCallback(firstNonNull(primary.batchCallback(), secondary.batchCallback()))
- .topics(primary.topics().isEmpty() ? secondary.topics() : primary.topics())
- .consumers(primary.consumers() != CONSUMERS_NOT_SET ? primary.consumers() : secondary.consumers())
+ .topics(
+ primary.topics()
+ .isEmpty() ? secondary.topics() : primary.topics()
+ )
+ .consumers(decideConsumerThreads(primary, secondary))
  .recoverCallback(firstNonNull(primary.recoverCallback(), secondary.recoverCallback()))
  .retryPolicy(firstNonNull(primary.retryPolicy(), secondary.retryPolicy()))
  .consumerSupplier(firstNonNull(primary.consumerSupplier(), secondary.consumerSupplier()))
  .pollInterval(firstNonNull(primary.pollInterval(), secondary.pollInterval()))
  .pollTimeout(firstNonNull(primary.pollTimeout(), secondary.pollTimeout()));
  }
 
- private static void fillWith(ConsumerConfig<?, ?> source, Builder target) {
+ private static <K, V> int decideConsumerThreads(ConsumerConfig<K, V> primary, ConsumerConfig<K, V> secondary) {
+ return primary.consumers() != CONSUMERS_NOT_SET ? primary.consumers() : secondary.consumers();
+ }
+
+ private static <K, V> void fillPropsWith(ConsumerConfig<K, V> source, Builder<K, V> target) {
  source
  .props()
  .forEach(target::prop);

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

@@ -14,6 +14,7 @@
 import lombok.SneakyThrows;
 import lombok.extern.slf4j.Slf4j;
 
+import static com.mageddo.kafka.client.ConsumerConfigDefault.CONSUMERS_NOT_SET;
 import static com.mageddo.kafka.client.DefaultConsumingConfig.DEFAULT_RETRY_STRATEGY;
 import static org.apache.kafka.clients.CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
 import static org.apache.kafka.clients.consumer.ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG;
@@ -60,7 +61,10 @@ public void consume(ConsumerConfig<K, V> consumerConfig) {
  }
  this.started = true;
  this.consumerConfig = consumerConfig;
- if (consumerConfig.consumers() == Integer.MIN_VALUE) {
+ final int threads = consumerConfig.consumers() != CONSUMERS_NOT_SET
+ ? consumerConfig.consumers()
+ : 1;
+ if (threads == Integer.MIN_VALUE) {
  log.info(
  "status=disabled-consumer, groupId={}, topics={}",
  this.consumerConfig.groupId(),
@@ -70,7 +74,7 @@ public void consume(ConsumerConfig<K, V> consumerConfig) {
  }
  this.checkReasonablePollInterval(consumerConfig);
 
- for (int i = 0; i < consumerConfig.consumers(); i++) {
+ for (int i = 0; i < threads; i++) {
  final ThreadConsumer<K, V> consumer = this.getInstance(this.create(consumerConfig), consumerConfig);
  consumer.start();
  }

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

@@ -52,43 +52,45 @@
  * }
  * </pre>
  */
-public class ConsumerStarter {
+public class ConsumerStarter<K, V> {
 
  private static final Logger log = LoggerFactory.getLogger(ConsumerStarter.class);
 
- private final ConsumerConfig<?, ?> config;
+ private final ConsumerConfig<K, V> config;
  private final List<ConsumerController<?, ?>> factories;
  private boolean started = false;
  private boolean stopped = false;
 
- public ConsumerStarter(ConsumerConfig<?, ?> config) {
+ public ConsumerStarter(ConsumerConfig<K, V> config) {
  this.config = config;
  this.factories = new ArrayList<>();
  }
 
- public static ConsumerStarter startFromConfig(ConsumerConfig<?, ?> config, List<ConsumerConfig> configs) {
- return new ConsumerStarter(config).startFromConfig(configs);
+ public static <K, V> ConsumerStarter<K, V> startFromConfig(
+ ConsumerConfig<K, V> config, List<ConsumerConfig<K, V>> configs
+ ) {
+ return new ConsumerStarter<K, V>(config).startFromConfig(configs);
  }
 
- public static ConsumerStarter start(ConsumerConfig<?, ?> config, List<Consumer> consumers) {
- return new ConsumerStarter(config).start(consumers);
+ public static <K, V> ConsumerStarter<K, V> start(ConsumerConfig<K, V> config, List<Consumer> consumers) {
+ return new ConsumerStarter<K, V>(config).start(consumers);
  }
 
- public ConsumerStarter start(List<Consumer> consumers) {
+ public ConsumerStarter<K, V> start(List<Consumer> consumers) {
  this.startFromConfig(consumers
  .stream()
- .map(Consumer::config)
+ .map(it -> (ConsumerConfig<K, V>) it.config())
  .collect(Collectors.toList())
  );
  return this;
  }
 
- public ConsumerStarter startFromConfig(List<ConsumerConfig> consumers) {
+ public ConsumerStarter<K, V> startFromConfig(List<ConsumerConfig<K, V>> consumers) {
  if (this.started) {
  throw new IllegalStateException("ConsumerConfig were already started");
  }
  this.started = true;
- for (final ConsumerConfig config : consumers) {
+ for (final ConsumerConfig<K, V> config : consumers) {
  this.factories.add(this.start(this.buildConsumer(config)));
  }
  return this;
@@ -128,16 +130,16 @@ public void stop() {
  }
  }
 
- public void waitFor(){
+ public void waitFor() {
  ConsumerConfigDefault.waitFor();
  }
 
- private ConsumerConfigDefault<?, ?> buildConsumer(ConsumerConfig<?, ?> config) {
+ private ConsumerConfigDefault<K, V> buildConsumer(ConsumerConfig<K, V> config) {
  return ConsumerConfigDefault.copy(config, this.config);
  }
 
- ConsumerController<?, ?> start(ConsumerConfig<?, ?> consumerConfig) {
- final ConsumerController consumerController = new ConsumerController<>();
+ ConsumerController<K,V> start(ConsumerConfig<K,V> consumerConfig) {
+ final ConsumerController<K,V> consumerController = new ConsumerController<>();
  consumerController.consume(consumerConfig);
  return consumerController;
  }

src/test/java/com/mageddo/kafka/client/ConsumerControllerTest.java

@@ -3,6 +3,7 @@
 import java.time.Duration;
 import java.util.Collections;
 
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.MockConsumer;
 import org.apache.kafka.clients.consumer.OffsetResetStrategy;
 import org.apache.kafka.common.config.ConfigException;
@@ -23,6 +24,7 @@
 import static org.mockito.ArgumentMatchers.eq;
 import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.never;
 import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
@@ -182,4 +184,28 @@ void mustUseDefaultPollTimeoutWhenItsNotSet() {
 
  }
 
+ @Test
+ @SneakyThrows
+ void mustUseDefaultConsumerThreadsWhenItsNotSet() {
+ // arrange
+ final var consumerConfig = ConsumerConfig
+ .<String, byte[]>builder()
+ .callback((callbackContext, record) -> System.out.println("nop"))
+ .build();
+
+ doReturn(mock(Consumer.class))
+ .when(this.consumerController)
+ .create(any());
+ doReturn(mock(ThreadConsumer.class))
+ .when(this.consumerController)
+ .getInstance(any(), any());
+
+ // act
+ this.consumerController.consume(consumerConfig);
+
+ // assert
+ verify(this.consumerController).getInstance(any(), any());
+
+ }
+
 }

src/test/java/com/mageddo/kafka/client/ConsumerStarterTest.java

@@ -77,7 +77,7 @@ void mustStartConsumerUsingTemplateConfigs() {
  .pollTimeout(templatePollTimeout)
  .retryPolicy(DEFAULT_RETRY_STRATEGY)
  .build();
- final var consumerStarter = spy(new ConsumerStarter(templateConfig));
+ final var consumerStarter = spy(new ConsumerStarter<>(templateConfig));
 
  final var consumerConfig = ConsumerConfig
  .builder()
@@ -101,18 +101,16 @@ void mustStartConsumerUsingTemplateConfigs() {
  assertEquals(templatePollInterval, actualConfig.pollInterval());
  assertEquals(templatePollTimeout, actualConfig.pollTimeout());
  assertEquals(1, actualConfig.props().size());
-
  }
 
-
  @Test
  void consumerSpecificConfigMustOverrideTemplateConfigs() {
 
  // arrange
  final var templateConfig = ConsumerConfigTemplates.builder()
  .prop(GROUP_ID_CONFIG, "my_group_id")
  .build();
- final var consumerStarter = spy(new ConsumerStarter(templateConfig));
+ final var consumerStarter = spy(new ConsumerStarter<>(templateConfig));
 
  final var groupId = "customGroupId";
  final var pollInterval = Duration.ofSeconds(33);
@@ -126,6 +124,7 @@ void consumerSpecificConfigMustOverrideTemplateConfigs() {
  .pollInterval(pollInterval)
  .pollTimeout(pollTimeout)
  .build();
+
  doReturn(mock(ConsumerController.class))
  .when(consumerStarter)
  .start(any(ConsumerConfig.class));
@@ -148,4 +147,32 @@ void consumerSpecificConfigMustOverrideTemplateConfigs() {
 
  }
 
+ @Test
+ void mustNotOverrideConsumerThreadsWhenDefaultValueIsDisabled() {
+
+ // arrange
+ final var consumingDisabled = Integer.MIN_VALUE;
+ final var templateConfig = ConsumerConfigTemplates.builder()
+ .consumers(consumingDisabled)
+ .build();
+ final var consumerStarter = spy(new ConsumerStarter<>(templateConfig));
+ final var consumers = 333;
+ final var consumerConfig = ConsumerConfig
+ .builder()
+ .consumers(consumers)
+ .build();
+
+ doReturn(mock(ConsumerController.class))
+ .when(consumerStarter)
+ .start(any(ConsumerConfig.class));
+
+ // act
+ consumerStarter.startFromConfig(List.of(consumerConfig));
+
+ // assert
+ verify(consumerStarter).start(this.argumentCaptor.capture());
+ final var actualConfig = this.argumentCaptor.getValue();
+ assertEquals(consumingDisabled, actualConfig.consumers());
+
+ }
 }