From: Kai Moritz <kai@juplo.de>
Date: Sun, 10 Apr 2022 17:03:38 +0000 (+0200)
Subject: Merge branch 'endless-stream-consumer' into rebalance-listener
X-Git-Tag: deserialization-synchroner-test~17
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=dd41cf0d4a8b1cf00158dd5ce902bba1ab67389b;p=demos%2Fkafka%2Ftraining

Merge branch 'endless-stream-consumer' into rebalance-listener
---

dd41cf0d4a8b1cf00158dd5ce902bba1ab67389b
diff --cc src/main/java/de/juplo/kafka/EndlessConsumer.java
index 14a875b,18cef23..03a89dc
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@@ -11,11 -9,13 +11,12 @@@ import org.apache.kafka.common.serializ
  
  import javax.annotation.PreDestroy;
  import java.time.Duration;
 -import java.util.Arrays;
 -import java.util.Properties;
 +import java.util.*;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
- import java.util.concurrent.Future;
- import java.util.concurrent.atomic.AtomicBoolean;
+ import java.util.concurrent.locks.Condition;
+ import java.util.concurrent.locks.Lock;
+ import java.util.concurrent.locks.ReentrantLock;
  
  
  @Slf4j
@@@ -28,14 -28,13 +29,16 @@@ public class EndlessConsumer implement
    private final String topic;
    private final String autoOffsetReset;
  
-   private AtomicBoolean running = new AtomicBoolean();
+   private final Lock lock = new ReentrantLock();
+   private final Condition condition = lock.newCondition();
+   private boolean running = false;
    private long consumed = 0;
    private KafkaConsumer<String, String> consumer = null;
-   private Future<?> future = null;
+ 
  
 +  private final Map<Integer, Map<String, Integer>> seen = new HashMap<>();
 +
 +
    public EndlessConsumer(
        ExecutorService executor,
        String bootstrapServer,
@@@ -151,19 -109,36 +155,41 @@@
      }
    }
  
+   private void shutdown()
+   {
+     lock.lock();
+     try
+     {
+       running = false;
+       condition.signal();
+     }
+     finally
+     {
+       lock.unlock();
+     }
+   }
+ 
 +  public Map<Integer, Map<String, Integer>> getSeen()
 +  {
 +    return seen;
 +  }
 +
-   public synchronized void start()
+   public void start()
    {
-     boolean stateChanged = running.compareAndSet(false, true);
-     if (!stateChanged)
-       throw new RuntimeException("Consumer instance " + id + " is already running!");
+     lock.lock();
+     try
+     {
+       if (running)
+         throw new IllegalStateException("Consumer instance " + id + " is already running!");
  
-     log.info("{} - Starting - consumed {} messages before", id, consumed);
-     future = executor.submit(this);
+       log.info("{} - Starting - consumed {} messages before", id, consumed);
+       running = true;
+       executor.submit(this);
+     }
+     finally
+     {
+       lock.unlock();
+     }
    }
  
    public synchronized void stop() throws ExecutionException, InterruptedException