Consumer Unsubscribes

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();
 }

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();

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()");

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 =

src/test/java/io/github/eocqrs/kafka/fake/FkConsumerTest.java

@@ -76,6 +76,10 @@ public void onPartitionsAssigned(final Collection<TopicPartition> partitions) {
       UnsupportedOperationException.class,
       () -> consumer.iterate("123", Duration.ofMillis(100L))
     );
+    assertThrows(
+      UnsupportedOperationException.class,
+      consumer::unsubscribe
+    );
     assertThrows(
       UnsupportedOperationException.class,
       consumer::close