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

diff --combined docker-compose.yml
index 0b1f0ae,b03bb1e..1392ae2
--- 1/docker-compose.yml
--- 2/docker-compose.yml
+++ b/docker-compose.yml
@@@ -24,13 -24,13 +24,13 @@@ services
      depends_on:
        - zookeeper
  
 -  setup:
 -    image: juplo/toolbox
 -    command: >
 -      bash -c "
 -        kafka-topics --bootstrap-server kafka:9092 --delete --if-exists --topic test
 -        kafka-topics --bootstrap-server kafka:9092 --create --topic test --partitions 2
 -      "
 +  kafka-ui:
 +    image: provectuslabs/kafka-ui:0.3.3
 +    ports:
 +      - 8080:8080
 +    environment:
 +      KAFKA_CLUSTERS_0_NAME: local
 +      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
  
    cli:
      image: juplo/toolbox
@@@ -39,18 -39,18 +39,18 @@@
    producer:
      image: juplo/endless-producer:1.0-SNAPSHOT
      ports:
-       - 8000:8080
+       - 8080:8880
      environment:
        producer.bootstrap-server: kafka:9092
        producer.client-id: producer
        producer.topic: test
 -      producer.throttle-ms: 200
 +      producer.throttle-ms: 10
  
  
    consumer:
      image: juplo/endless-consumer:1.0-SNAPSHOT
      ports:
-       - 8081:8081
+       - 8081:8881
      environment:
        consumer.bootstrap-server: kafka:9092
        consumer.client-id: my-group

diff --combined pom.xml
index d9b70f9,cf68030..8c5dccc
--- 1/pom.xml
--- 2/pom.xml
+++ b/pom.xml
@@@ -14,7 -14,7 +14,7 @@@
    <groupId>de.juplo.kafka</groupId>
    <artifactId>endless-consumer</artifactId>
    <version>1.0-SNAPSHOT</version>
 -  <name>Endless Consumer: a Simple Consumer-Group that reads and print the topic</name>
 +  <name>Endless Consumer: a Simple Consumer-Group that reads and prints the topic and counts the received messages for each key by topic</name>
  
    <dependencies>
      <dependency>
@@@ -54,6 -54,17 +54,17 @@@
        <plugin>
          <groupId>org.springframework.boot</groupId>
          <artifactId>spring-boot-maven-plugin</artifactId>
+         <executions>
+           <execution>
+             <goals>
+               <goal>build-info</goal>
+             </goals>
+           </execution>
+         </executions>
+       </plugin>
+       <plugin>
+         <groupId>pl.project13.maven</groupId>
+         <artifactId>git-commit-id-plugin</artifactId>
        </plugin>
        <plugin>
          <groupId>io.fabric8</groupId>

diff --combined src/main/java/de/juplo/kafka/ApplicationHealthIndicator.java
index 0000000,0000000..ab9782c
new file mode 100644 (file)
--- /dev/null
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/ApplicationHealthIndicator.java
@@@ -1,0 -1,0 +1,32 @@@
++package de.juplo.kafka;
++
++import lombok.RequiredArgsConstructor;
++import org.springframework.boot.actuate.health.Health;
++import org.springframework.boot.actuate.health.HealthIndicator;
++import org.springframework.stereotype.Component;
++
++
++@Component
++@RequiredArgsConstructor
++public class ApplicationHealthIndicator implements HealthIndicator
++{
++  private final EndlessConsumer consumer;
++
++
++  @Override
++  public Health health()
++  {
++    try
++    {
++      return consumer
++          .exitStatus()
++          .map(Health::down)
++          .orElse(Health.outOfService())
++          .build();
++    }
++    catch (IllegalStateException e)
++    {
++      return Health.up().build();
++    }
++  }
++}

diff --combined src/main/java/de/juplo/kafka/DriverController.java
index a504842,a02fd2c..1525f5a
--- 1/src/main/java/de/juplo/kafka/DriverController.java
--- 2/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.http.HttpStatus;
+ import org.springframework.web.bind.annotation.ExceptionHandler;
 +import org.springframework.web.bind.annotation.GetMapping;
  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,16 @@@ 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 --combined src/main/java/de/juplo/kafka/EndlessConsumer.java
index 03a89dc,adebff1..c2d4447
--- 1/src/main/java/de/juplo/kafka/EndlessConsumer.java
--- 2/src/main/java/de/juplo/kafka/EndlessConsumer.java
+++ b/src/main/java/de/juplo/kafka/EndlessConsumer.java
@@@ -1,17 -1,17 +1,17 @@@
  package de.juplo.kafka;
  
  import lombok.extern.slf4j.Slf4j;
 +import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
  import org.apache.kafka.clients.consumer.ConsumerRecord;
  import org.apache.kafka.clients.consumer.ConsumerRecords;
  import org.apache.kafka.clients.consumer.KafkaConsumer;
 +import org.apache.kafka.common.TopicPartition;
  import org.apache.kafka.common.errors.WakeupException;
  import org.apache.kafka.common.serialization.StringDeserializer;
  
  import javax.annotation.PreDestroy;
  import java.time.Duration;
 -import java.util.Arrays;
 -import java.util.Optional;
 -import java.util.Properties;
 +import java.util.*;
  import java.util.concurrent.ExecutionException;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.locks.Condition;
@@@ -32,13 -32,11 +32,14 @@@ public class EndlessConsumer implement
    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 final Map<Integer, Map<String, Integer>> seen = new HashMap<>();
 +
 +
    public EndlessConsumer(
        ExecutorService executor,
        String bootstrapServer,
@@@ -65,44 -63,13 +66,44 @@@
        props.put("group.id", groupId);
        props.put("client.id", id);
        props.put("auto.offset.reset", autoOffsetReset);
 +      props.put("metadata.max.age.ms", "1000");
        props.put("key.deserializer", StringDeserializer.class.getName());
        props.put("value.deserializer", StringDeserializer.class.getName());
  
        this.consumer = new KafkaConsumer<>(props);
  
        log.info("{} - Subscribing to topic {}", id, topic);
 -      consumer.subscribe(Arrays.asList(topic));
 +      consumer.subscribe(Arrays.asList(topic), new ConsumerRebalanceListener()
 +      {
 +        @Override
 +        public void onPartitionsRevoked(Collection<TopicPartition> partitions)
 +        {
 +          partitions.forEach(tp ->
 +          {
 +            log.info("{} - removing partition: {}", id, tp);
 +            Map<String, Integer> removed = seen.remove(tp.partition());
 +            for (String key : removed.keySet())
 +            {
 +              log.info(
 +                  "{} - Seen {} messages for partition={}|key={}",
 +                  id,
 +                  removed.get(key),
 +                  tp.partition(),
 +                  key);
 +            }
 +          });
 +        }
 +
 +        @Override
 +        public void onPartitionsAssigned(Collection<TopicPartition> partitions)
 +        {
 +          partitions.forEach(tp ->
 +          {
 +            log.info("{} - adding partition: {}", id, tp);
 +            seen.put(tp.partition(), new HashMap<>());
 +          });
 +        }
 +      });
  
        while (true)
        {
@@@ -123,17 -90,6 +124,17 @@@
                record.key(),
                record.value()
            );
 +
 +          Integer partition = record.partition();
 +          String key = record.key() == null ? "NULL" : record.key();
 +          Map<String, Integer> byKey = seen.get(partition);
 +
 +          if (!byKey.containsKey(key))
 +            byKey.put(key, 0);
 +
 +          int seenByKey = byKey.get(key);
 +          seenByKey++;
 +          byKey.put(key, seenByKey);
          }
        }
      }
@@@ -145,7 -101,7 +146,7 @@@
      catch(Exception e)
      {
        log.error("{} - Unexpected error: {}", id, e.toString(), e);
-       shutdown();
+       shutdown(e);
      }
      finally
      {
@@@ -156,11 -112,17 +157,17 @@@
    }
  
    private void shutdown()
+   {
+     shutdown(null);
+   }
+ 
+   private void shutdown(Exception e)
    {
      lock.lock();
      try
      {
        running = false;
+       exception = e;
        condition.signal();
      }
      finally
@@@ -169,11 -131,6 +176,11 @@@
      }
    }
  
 +  public Map<Integer, Map<String, Integer>> getSeen()
 +  {
 +    return seen;
 +  }
 +
    public void start()
    {
      lock.lock();
@@@ -184,6 -141,7 +191,7 @@@
  
        log.info("{} - Starting - consumed {} messages before", id, consumed);
        running = true;
+       exception = null;
        executor.submit(this);
      }
      finally
@@@ -232,4 -190,33 +240,33 @@@
        log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
      }
    }
+ 
+   public boolean running()
+   {
+     lock.lock();
+     try
+     {
+       return running;
+     }
+     finally
+     {
+       lock.unlock();
+     }
+   }
+ 
+   public Optional<Exception> exitStatus()
+   {
+     lock.lock();
+     try
+     {
+       if (running)
+         throw new IllegalStateException("No exit-status available: Consumer instance " + id + " is running!");
+ 
+       return Optional.ofNullable(exception);
+     }
+     finally
+     {
+       lock.unlock();
+     }
+   }
  }