action poc

src/main/java/io/github/eocqrs/kafka/ext/Action.java

@@ -0,0 +1,10 @@
+package io.github.eocqrs.kafka.ext;
+
+/**
+ * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
+ * @since 0.2.5
+ */
+public interface Action {
+
+  void apply();
+}

src/main/java/io/github/eocqrs/kafka/ext/AssignPartitions.java

@@ -0,0 +1,30 @@
+package io.github.eocqrs.kafka.ext;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.TopicPartition;
+import org.cactoos.list.ListOf;
+
+import java.util.Collection;
+
+/**
+ * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
+ * @since 0.2.5
+ */
+public final class AssignPartitions implements Action {
+
+  private final KafkaConsumer<?, ?> consumer;
+  private final Collection<TopicPartition> partitions;
+
+  public AssignPartitions(
+    final KafkaConsumer<?, ?> cnsmr,
+    TopicPartition... prts
+  ) {
+    this.consumer = cnsmr;
+    this.partitions = new ListOf<>(prts);
+  }
+
+  @Override
+  public void apply() {
+    this.consumer.assign(this.partitions);
+  }
+}

src/main/java/io/github/eocqrs/kafka/ext/CommitAsync.java

@@ -0,0 +1,19 @@
+package io.github.eocqrs.kafka.ext;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+
+/**
+ * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
+ * @since 0.2.5
+ */
+@RequiredArgsConstructor
+public final class CommitAsync implements Action {
+
+  private final KafkaConsumer<?, ?> consumer;
+
+  @Override
+  public void apply() {
+    this.consumer.commitAsync();
+  }
+}

src/main/java/io/github/eocqrs/kafka/ext/CommitSync.java

@@ -0,0 +1,19 @@
+package io.github.eocqrs.kafka.ext;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+
+/**
+ * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
+ * @since 0.2.5
+ */
+@RequiredArgsConstructor
+public final class CommitSync implements Action {
+
+  private final KafkaConsumer<?, ?> consumer;
+
+  @Override
+  public void apply() {
+    this.consumer.commitSync();
+  }
+}

src/main/java/io/github/eocqrs/kafka/ext/ShutdownHook.java

@@ -0,0 +1,23 @@
+package io.github.eocqrs.kafka.ext;
+
+import lombok.RequiredArgsConstructor;
+
+/**
+ * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
+ * @since 0.2.5
+ */
+@RequiredArgsConstructor
+public final class ShutdownHook implements Action {
+
+  /**
+   * Thread.
+   */
+  private final Thread thread;
+
+  @Override
+  public void apply() {
+    Runtime.getRuntime().addShutdownHook(
+      this.thread
+    );
+  }
+}

src/main/java/io/github/eocqrs/kafka/ext/Wakeup.java

@@ -0,0 +1,19 @@
+package io.github.eocqrs.kafka.ext;
+
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+
+/**
+ * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
+ * @since 0.2.5
+ */
+@RequiredArgsConstructor
+public final class Wakeup implements Action {
+
+  private final KafkaConsumer<?, ?> consumer;
+
+  @Override
+  public void apply() {
+    this.consumer.wakeup();
+  }
+}