Refactor: Logik für Counter in Klasse `CounterState` extrahiert

diff --git a/src/main/java/de/juplo/kafka/CounterState.java b/src/main/java/de/juplo/kafka/CounterState.java
new file mode 100644 (file)
index 0000000..5e39371
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/CounterState.java
@@ -0,0 +1,26 @@
+package de.juplo.kafka;
+
+import java.util.HashMap;
+import java.util.Map;
+
+
+public class CounterState
+{
+  private final Map<String, Long> counterState = new HashMap<>();
+
+
+  public void setCounterState(String key, long counter)
+  {
+    counterState.put(key, counter);
+  }
+
+  public synchronized Long addToCounter(String key)
+  {
+    return counterState.compute(key, (k, v) -> v == null ? 1l : v + 1);
+  }
+
+  public synchronized Map<String, Long> getCounterState()
+  {
+    return counterState;
+  }
+}

diff --git a/src/main/java/de/juplo/kafka/ExampleConsumer.java b/src/main/java/de/juplo/kafka/ExampleConsumer.java
index 90402f8..0b57140 100644 (file)
--- a/src/main/java/de/juplo/kafka/ExampleConsumer.java
+++ b/src/main/java/de/juplo/kafka/ExampleConsumer.java
@@ -24,7 +24,7 @@ public class ExampleConsumer implements Runnable, ConsumerRebalanceListener
   private final Thread workerThread;
   private final Runnable closeCallback;
 
-  private final Map<String, Long> counterState = new HashMap<>();
+  private final CounterState counterState = new CounterState();
   private final String stateTopic;
   private final Producer<String, String> producer;
 
@@ -174,7 +174,7 @@ public class ExampleConsumer implements Runnable, ConsumerRebalanceListener
     String key,
     String value)
   {
-    counterState.put(key, Long.parseLong(value));
+    counterState.setCounterState(key, Long.parseLong(value));
     if (offset + 1 == stateEndOffsets[partition])
     {
       log.info(
@@ -214,12 +214,12 @@ public class ExampleConsumer implements Runnable, ConsumerRebalanceListener
 
   private synchronized Long computeCount(String key)
   {
-    return counterState.compute(key, (k, v) -> v == null ? 1l : v + 1);
+    return counterState.addToCounter(key);
   }
 
-  public synchronized Map<String, Long> getCounterState()
+  public Map<String, Long> getCounterState()
   {
-    return counterState;
+    return counterState.getCounterState();
   }
 
   void sendCounterState(int partition, String key, Long counter)