#288

eo-cqrs · May 4, 2023 · 3840661 · 3840661
1 parent a85fa7f
commit 3840661
Show file tree

Hide file tree

Showing 4 changed files with 31 additions and 1 deletion.
diff --git a/src/main/java/io/github/eocqrs/kafka/Consumer.java b/src/main/java/io/github/eocqrs/kafka/Consumer.java
@@ -26,6 +26,7 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.List;
+
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 
 /**
@@ -69,4 +70,9 @@ public interface Consumer<K, X> extends Closeable {
    * @return Dataized polled data.
    */
   List<Dataized<X>> iterate(String topic, Duration timeout);
+
+  /**
+   * Unsubscribe.
+   */
+  void unsubscribe();
 }
diff --git a/src/main/java/io/github/eocqrs/kafka/consumer/KfConsumer.java b/src/main/java/io/github/eocqrs/kafka/consumer/KfConsumer.java
@@ -110,6 +110,11 @@ public List<Dataized<X>> iterate(final String topic, final Duration timeout) {
     return accumulator;
   }
 
+  @Override
+  public void unsubscribe() {
+    this.origin.unsubscribe();
+  }
+
   @Override
   public void close() {
     this.origin.close();

diff --git a/src/main/java/io/github/eocqrs/kafka/fake/FkConsumer.java b/src/main/java/io/github/eocqrs/kafka/fake/FkConsumer.java
@@ -64,6 +64,11 @@ public List<Dataized<X>> iterate(final String topic, final Duration timeout) {
     throw new UnsupportedOperationException("#iterate()");
   }
 
+  @Override
+  public void unsubscribe() {
+    throw new UnsupportedOperationException("#unsubscribe()");
+  }
+
   @Override
   public void close() {
     throw new UnsupportedOperationException("#close()");

diff --git a/src/test/java/io/github/eocqrs/kafka/consumer/KfConsumerTest.java b/src/test/java/io/github/eocqrs/kafka/consumer/KfConsumerTest.java
@@ -32,7 +32,9 @@
 import io.github.eocqrs.kafka.parameters.KfParams;
 import io.github.eocqrs.kafka.parameters.ValueDeserializer;
 import io.github.eocqrs.kafka.xml.KfXmlFlexible;
+
 import java.util.Collection;
+
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
@@ -55,7 +57,7 @@
 final class KfConsumerTest {
 
   @Test
-  void subscribes(
+  void subscribesWithoutException(
     @Mock final ConsumerSettings<String, String> settings,
     @Mock final KafkaConsumer<String, String> consumer
   ) {
@@ -81,6 +83,18 @@ public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
     );
   }
 
+  @Test
+  void unsubscribesWithoutException(
+    @Mock final ConsumerSettings<String, String> settings,
+    @Mock final KafkaConsumer<String, String> origin
+  ) {
+    Mockito.when(settings.consumer()).thenReturn(origin);
+    final Consumer<String, String> consumer =
+      new KfConsumer<>(settings);
+    assertDoesNotThrow(consumer::unsubscribe);
+    assertDoesNotThrow(consumer::close);
+  }
+
   @Test
   void constructsConsumerWithXML() throws Exception {
     final Consumer<String, String> consumer =