From: Kai Moritz <kai@juplo.de>
Date: Sat, 23 Jul 2022 11:49:23 +0000 (+0200)
Subject: Merge der überarbeiteten Compose-Konfiguration ('endless-stream-consumer')
X-Git-Tag: sumup-requests---lvm-2-tage~5^2~1
X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;ds=sidebyside;h=f9c0ba7779552d8fcfc9cb29c8b689e20c314904;p=demos%2Fkafka%2Ftraining

Merge der überarbeiteten Compose-Konfiguration ('endless-stream-consumer')

* Die letzten Änderungen an 'endless-stream-consumer' sind länger nicht
  mehr gemerged worden.
---

f9c0ba7779552d8fcfc9cb29c8b689e20c314904
diff --cc docker-compose.yml
index 0b1f0ae,159f9cb..f9eeedd
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@@ -39,8 -39,9 +39,9 @@@ services
    producer:
      image: juplo/endless-producer:1.0-SNAPSHOT
      ports:
 -      - 8080:8080
 +      - 8000:8080
      environment:
+       server.port: 8080
        producer.bootstrap-server: kafka:9092
        producer.client-id: producer
        producer.topic: test
diff --cc src/main/java/de/juplo/kafka/DriverController.java
index a504842,a02fd2c..06e562c
--- a/src/main/java/de/juplo/kafka/DriverController.java
+++ b/src/main/java/de/juplo/kafka/DriverController.java
@@@ -1,11 -1,12 +1,14 @@@
  package de.juplo.kafka;
  
  import lombok.RequiredArgsConstructor;
 +import org.springframework.web.bind.annotation.GetMapping;
+ import org.springframework.http.HttpStatus;
+ import org.springframework.web.bind.annotation.ExceptionHandler;
  import org.springframework.web.bind.annotation.PostMapping;
+ import org.springframework.web.bind.annotation.ResponseStatus;
  import org.springframework.web.bind.annotation.RestController;
  
 +import java.util.Map;
  import java.util.concurrent.ExecutionException;
  
  
@@@ -28,10 -29,10 +31,18 @@@ public class DriverControlle
      consumer.stop();
    }
  
 +
 +  @GetMapping("seen")
 +  public Map<Integer, Map<String, Integer>> seen()
 +  {
 +    return consumer.getSeen();
 +  }
++
++
+   @ExceptionHandler
+   @ResponseStatus(HttpStatus.BAD_REQUEST)
+   public ErrorResponse illegalStateException(IllegalStateException e)
+   {
+     return new ErrorResponse(e.getMessage(), HttpStatus.BAD_REQUEST.value());
+   }
  }
diff --cc src/main/java/de/juplo/kafka/EndlessConsumer.java
index 357a0b4,adebff1..2310ccd
--- a/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@@ -10,8 -10,7 +10,9 @@@ import org.apache.kafka.common.serializ
  import javax.annotation.PreDestroy;
  import java.time.Duration;
  import java.util.Arrays;
 +import java.util.HashMap;
 +import java.util.Map;
+ import java.util.Optional;
  import java.util.Properties;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
@@@ -29,14 -29,14 +31,17 @@@ 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 Exception exception;
    private long consumed = 0;
    private KafkaConsumer<String, String> consumer = null;
-   private Future<?> future = null;
+ 
  
 +  private Map<Integer, Map<String, Integer>> seen;
 +
 +
    public EndlessConsumer(
        ExecutorService executor,
        String bootstrapServer,
@@@ -144,19 -111,43 +150,48 @@@
      }
    }
  
 +  public Map<Integer, Map<String, Integer>> getSeen()
 +  {
 +    return seen;
 +  }
 +
-   public synchronized void start()
+   private void shutdown()
+   {
+     shutdown(null);
+   }
+ 
+   private void shutdown(Exception e)
+   {
+     lock.lock();
+     try
+     {
+       running = false;
+       exception = e;
+       condition.signal();
+     }
+     finally
+     {
+       lock.unlock();
+     }
+   }
+ 
+   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;
+       exception = null;
+       executor.submit(this);
+     }
+     finally
+     {
+       lock.unlock();
+     }
    }
  
    public synchronized void stop() throws ExecutionException, InterruptedException