Make Kafka producer/consumer testable

Refactor KafkaMessageProducer and KafkaMessageConsumer to depend on the Producer/Consumer interfaces and add constructors that accept mockable instances. Extract default producer/consumer creation into factory methods so tests can inject MockProducer/MockConsumer. Update tests across the microservices-messaging module to use MockProducer/MockConsumer, add more meaningful assertions, error/interrupt handling tests, and simplify AppTest. These changes improve unit testability and remove the need for a running Kafka instance while preserving runtime behavior.

Author: Mukul-Howale

microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageConsumer.java

@@ -30,7 +30,7 @@
 import java.util.Collections;
 import java.util.Properties;
 import java.util.concurrent.atomic.AtomicBoolean;
-import java.util.function.Consumer;
+import org.apache.kafka.clients.consumer.Consumer;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -41,10 +41,10 @@
 /** Kafka message consumer that subscribes to topics and processes messages. */
 public class KafkaMessageConsumer implements AutoCloseable, Runnable {
   private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageConsumer.class);
-  private final KafkaConsumer<String, String> consumer;
+  private final Consumer<String, String> consumer;
   private final ObjectMapper objectMapper;
   private final String topic;
-  private final Consumer<Message> messageHandler;
+  private final java.util.function.Consumer<Message> messageHandler;
   private final AtomicBoolean running = new AtomicBoolean(true);
 
   /**
@@ -56,20 +56,34 @@ public class KafkaMessageConsumer implements AutoCloseable, Runnable {
    * @param messageHandler handler for received messages
    */
   public KafkaMessageConsumer(
-      String bootstrapServers, String groupId, String topic, Consumer<Message> messageHandler) {
+      String bootstrapServers,
+      String groupId,
+      String topic,
+      java.util.function.Consumer<Message> messageHandler) {
+    this(createDefaultConsumer(bootstrapServers, groupId), topic, messageHandler);
+  }
+
+  KafkaMessageConsumer(
+      Consumer<String, String> consumer,
+      String topic,
+      java.util.function.Consumer<Message> messageHandler) {
+    this.consumer = consumer;
+    this.objectMapper = new ObjectMapper();
+    this.objectMapper.registerModule(new JavaTimeModule());
+    this.topic = topic;
+    this.messageHandler = messageHandler;
+  }
+
+  private static Consumer<String, String> createDefaultConsumer(
+      String bootstrapServers, String groupId) {
     Properties props = new Properties();
     props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
     props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
     props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
     props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class.getName());
     props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
     props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
-
-    this.consumer = new KafkaConsumer<>(props);
-    this.objectMapper = new ObjectMapper();
-    this.objectMapper.registerModule(new JavaTimeModule());
-    this.topic = topic;
-    this.messageHandler = messageHandler;
+    return new KafkaConsumer<>(props);
   }
 
   @Override

microservices-messaging/src/main/java/com/iluwatar/messaging/KafkaMessageProducer.java

@@ -28,6 +28,7 @@
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
 import java.util.Properties;
 import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.Producer;
 import org.apache.kafka.clients.producer.ProducerConfig;
 import org.apache.kafka.clients.producer.ProducerRecord;
 import org.apache.kafka.common.serialization.StringSerializer;
@@ -37,7 +38,7 @@
 /** Kafka message producer that publishes messages to Kafka topics. */
 public class KafkaMessageProducer implements AutoCloseable {
   private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageProducer.class);
-  private final KafkaProducer<String, String> producer;
+  private final Producer<String, String> producer;
   private final ObjectMapper objectMapper;
 
   /**
@@ -46,16 +47,23 @@ public class KafkaMessageProducer implements AutoCloseable {
    * @param bootstrapServers Kafka bootstrap servers
    */
   public KafkaMessageProducer(String bootstrapServers) {
+    this(createDefaultProducer(bootstrapServers));
+  }
+
+  KafkaMessageProducer(Producer<String, String> producer) {
+    this.producer = producer;
+    this.objectMapper = new ObjectMapper();
+    this.objectMapper.registerModule(new JavaTimeModule());
+  }
+
+  private static Producer<String, String> createDefaultProducer(String bootstrapServers) {
     Properties props = new Properties();
     props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
     props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
     props.put(ProducerConfig.ACKS_CONFIG, "all");
     props.put(ProducerConfig.RETRIES_CONFIG, 3);
-
-    this.producer = new KafkaProducer<>(props);
-    this.objectMapper = new ObjectMapper();
-    this.objectMapper.registerModule(new JavaTimeModule());
+    return new KafkaProducer<>(props);
   }
 
   /**

microservices-messaging/src/test/java/com/iluwatar/messaging/AppTest.java

@@ -32,17 +32,7 @@
 class AppTest {
 
   @Test
-  void testMainMethodDoesNotThrowException() {
-    // Note: This test requires a running Kafka instance
-    // In a real scenario, we would use embedded Kafka for testing
-    // For now, we just verify the method can be called without compilation errors
-
-    // Act & Assert
-    assertDoesNotThrow(
-        () -> {
-          // Main method requires Kafka to be running, so we don't actually call it in unit tests
-          // This is a placeholder to ensure the class structure is correct
-        },
-        "App should be instantiable");
+  void testAppConstructor() {
+    assertDoesNotThrow(() -> new App());
   }
 }

microservices-messaging/src/test/java/com/iluwatar/messaging/InventoryServiceTest.java

@@ -104,4 +104,15 @@ void testHandleMultipleMessages() {
         },
         "Should handle multiple messages without error");
   }
+
+  @Test
+  void testHandleMessagesWhenInterrupted() {
+    Thread.currentThread().interrupt();
+    inventoryService.handleMessage(new Message("Order Created: ORDER-001"));
+    org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
+
+    Thread.currentThread().interrupt();
+    inventoryService.handleMessage(new Message("Order Cancelled: ORDER-001"));
+    org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
+  }
 }

microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageConsumerTest.java

@@ -26,57 +26,92 @@
 
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.MockConsumer;
+import org.apache.kafka.clients.consumer.OffsetResetStrategy;
+import org.apache.kafka.common.TopicPartition;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 /**
- * Unit tests for {@link KafkaMessageConsumer}. Note: These tests verify basic functionality without
- * requiring a Kafka instance. For integration tests with Kafka, use embedded Kafka or
- * testcontainers.
+ * Unit tests for {@link KafkaMessageConsumer}.
  */
 class KafkaMessageConsumerTest {
 
+  private MockConsumer<String, String> mockConsumer;
+  private KafkaMessageConsumer kafkaMessageConsumer;
+  private AtomicBoolean handlerCalled;
+
+  @BeforeEach
+  void setUp() {
+    mockConsumer = new MockConsumer<>(OffsetResetStrategy.EARLIEST);
+    handlerCalled = new AtomicBoolean(false);
+    kafkaMessageConsumer = new KafkaMessageConsumer(
+        mockConsumer,
+        "test-topic",
+        msg -> handlerCalled.set(true));
+  }
+
   @Test
   void testConsumerCanBeInstantiated() {
-    // Arrange & Act & Assert
-    // Note: Don't actually create consumer in unit test as it requires Kafka
-    assertNotNull(KafkaMessageConsumer.class, "KafkaMessageConsumer class should exist");
+    assertNotNull(kafkaMessageConsumer, "KafkaMessageConsumer should be instantiated");
   }
 
   @Test
-  void testConsumerImplementsRunnable() {
-    // Arrange & Act & Assert
-    var interfaces = KafkaMessageConsumer.class.getInterfaces();
-    for (var i : interfaces) {
-      if (i.equals(Runnable.class)) {
-        break;
+  void testRunProcessesValidMessageAndStops() throws Exception {
+    ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule());
+    Message msg = new Message("Order Created: 123");
+    String jsonStr = mapper.writeValueAsString(msg);
+
+    TopicPartition tp = new TopicPartition("test-topic", 0);
+    mockConsumer.updateBeginningOffsets(new HashMap<>() {
+      {
+        put(tp, 0L);
       }
-    }
-    assertNotNull(interfaces, "Should have interfaces");
-    // Note: Runnable is implemented for threading
+    });
+
+    mockConsumer.schedulePollTask(() -> {
+      mockConsumer.rebalance(Collections.singletonList(tp));
+      mockConsumer.addRecord(new ConsumerRecord<>("test-topic", 0, 0L, "key", jsonStr));
+    });
+
+    mockConsumer.schedulePollTask(() -> kafkaMessageConsumer.stop());
+
+    kafkaMessageConsumer.run();
+
+    assertTrue(handlerCalled.get(), "Handler should have been invoked");
+    assertTrue(mockConsumer.closed(), "Consumer should be closed");
   }
 
   @Test
-  void testConsumerImplementsAutoCloseable() {
-    // Arrange & Act & Assert
-    var interfaces = KafkaMessageConsumer.class.getInterfaces();
-    for (var i : interfaces) {
-      if (i.equals(AutoCloseable.class)) {
-        break;
+  void testRunHandlesInvalidJsonMessage() {
+    TopicPartition tp = new TopicPartition("test-topic", 0);
+    mockConsumer.updateBeginningOffsets(new HashMap<>() {
+      {
+        put(tp, 0L);
       }
-    }
-    assertNotNull(interfaces, "Should have interfaces");
-    // Note: AutoCloseable is implemented
+    });
+
+    mockConsumer.schedulePollTask(() -> {
+      mockConsumer.rebalance(Collections.singletonList(tp));
+      mockConsumer.addRecord(new ConsumerRecord<>("test-topic", 0, 0L, "key", "{invalid json"));
+    });
+
+    mockConsumer.schedulePollTask(() -> kafkaMessageConsumer.stop());
+
+    assertDoesNotThrow(() -> kafkaMessageConsumer.run());
+    assertTrue(mockConsumer.closed(), "Consumer should be closed");
   }
 
   @Test
-  void testConsumerClassHasStopMethod() {
-    // Arrange & Act & Assert
-    assertDoesNotThrow(
-        () -> {
-          var method = KafkaMessageConsumer.class.getDeclaredMethod("stop");
-          assertNotNull(method, "stop method should exist");
-        },
-        "KafkaMessageConsumer should have stop method");
+  void testCloseStopsConsumer() {
+    assertDoesNotThrow(() -> kafkaMessageConsumer.close());
   }
 }

microservices-messaging/src/test/java/com/iluwatar/messaging/KafkaMessageProducerTest.java

@@ -24,48 +24,61 @@
  */
 package com.iluwatar.messaging;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import org.apache.kafka.clients.producer.MockProducer;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 /**
- * Unit tests for {@link KafkaMessageProducer}. Note: These tests verify basic functionality without
- * requiring a Kafka instance. For integration tests with Kafka, use embedded Kafka or
- * testcontainers.
+ * Unit tests for {@link KafkaMessageProducer}.
  */
 class KafkaMessageProducerTest {
 
+  private MockProducer<String, String> mockProducer;
+  private KafkaMessageProducer kafkaMessageProducer;
+
+  @BeforeEach
+  void setUp() {
+    mockProducer = new MockProducer<>(true, new StringSerializer(), new StringSerializer());
+    kafkaMessageProducer = new KafkaMessageProducer(mockProducer);
+  }
+
   @Test
   void testProducerCanBeInstantiated() {
-    // Arrange & Act & Assert
-    // Note: Don't actually create producer in unit test as it requires Kafka
-    // This test just verifies the class structure is correct
-    assertNotNull(KafkaMessageProducer.class, "KafkaMessageProducer class should exist");
+    assertNotNull(kafkaMessageProducer, "KafkaMessageProducer should be instantiated");
   }
 
   @Test
-  void testProducerClassHasPublishMethod() {
-    // Arrange & Act & Assert
-    assertDoesNotThrow(
-        () -> {
-          var method =
-              KafkaMessageProducer.class.getDeclaredMethod("publish", String.class, Message.class);
-          assertNotNull(method, "publish method should exist");
-        },
-        "KafkaMessageProducer should have publish method");
+  void testPublishMessageSuccess() {
+    Message message = new Message("Test Order");
+
+    assertDoesNotThrow(() -> kafkaMessageProducer.publish("test-topic", message));
+    assertEquals(1, mockProducer.history().size());
+    assertEquals("test-topic", mockProducer.history().get(0).topic());
+    assertEquals(message.getId(), mockProducer.history().get(0).key());
   }
 
   @Test
-  void testProducerImplementsAutoCloseable() {
-    // Arrange & Act & Assert
-    var interfaces = KafkaMessageProducer.class.getInterfaces();
-    for (var i : interfaces) {
-      if (i.equals(AutoCloseable.class)) {
-        break;
-      }
+  void testPublishMessageErrorCallback() {
+    MockProducer<String, String> failingProducer =
+        new MockProducer<>(false, new StringSerializer(), new StringSerializer());
+    try (KafkaMessageProducer producerWithError = new KafkaMessageProducer(failingProducer)) {
+      Message message = new Message("Test Order");
+
+      assertDoesNotThrow(() -> producerWithError.publish("test-topic", message));
+      assertEquals(1, failingProducer.history().size());
+      failingProducer.errorNext(new RuntimeException("Kafka publish error"));
     }
-    assertNotNull(interfaces, "Should have interfaces");
-    // Note: AutoCloseable is implemented
+  }
+
+  @Test
+  void testClose() {
+    assertDoesNotThrow(() -> kafkaMessageProducer.close());
+    assertTrue(mockProducer.closed());
   }
 }

microservices-messaging/src/test/java/com/iluwatar/messaging/NotificationServiceTest.java

@@ -116,4 +116,19 @@ void testHandleMultipleOrdersSequentially() {
         },
         "Should handle multiple orders sequentially without error");
   }
+
+  @Test
+  void testHandleMessagesWhenInterrupted() {
+    Thread.currentThread().interrupt();
+    notificationService.handleMessage(new Message("Order Created: ORDER-001"));
+    org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
+
+    Thread.currentThread().interrupt();
+    notificationService.handleMessage(new Message("Order Updated: ORDER-001"));
+    org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
+
+    Thread.currentThread().interrupt();
+    notificationService.handleMessage(new Message("Order Cancelled: ORDER-001"));
+    org.junit.jupiter.api.Assertions.assertTrue(Thread.interrupted());
+  }
 }