version: '3.2'
services:
zookeeper:
- image: confluentinc/cp-zookeeper:7.0.2
+ image: confluentinc/cp-zookeeper:7.1.3
environment:
ZOOKEEPER_CLIENT_PORT: 2181
ports:
- 2181:2181
kafka:
- image: confluentinc/cp-kafka:7.0.2
+ image: confluentinc/cp-kafka:7.1.3
environment:
KAFKA_BROKER_ID: 1
KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
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
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
- producer.throttle-ms: 200
+ producer.throttle-ms: 10
consumer:
image: juplo/endless-consumer:1.0-SNAPSHOT
ports:
- - 8081:8081
+ - 8081:8080
environment:
+ server.port: 8080
consumer.bootstrap-server: kafka:9092
consumer.client-id: my-group
consumer.client-id: consumer
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
- <version>2.6.5</version>
+ <version>2.7.2</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<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>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
+ <dependency>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-validation</artifactId>
+ </dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
<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>
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;
- 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
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,
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());
log.info("{} - Subscribing to topic {}", id, topic);
consumer.subscribe(Arrays.asList(topic));
+ seen = new HashMap<>();
+
while (true)
{
ConsumerRecords<String, String> records =
record.key(),
record.value()
);
+
+ Integer partition = record.partition();
+ String key = record.key() == null ? "NULL" : record.key();
+
+ if (!seen.containsKey(partition))
+ seen.put(partition, new HashMap<>());
+
+ 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);
}
}
}
catch(WakeupException e)
{
log.info("{} - RIIING!", id);
+ shutdown();
}
catch(Exception e)
{
log.error("{} - Unexpected error: {}", id, e.toString(), e);
- running.set(false); // Mark the instance as not running
+ shutdown(e);
}
finally
{
log.info("{} - Closing the KafkaConsumer", id);
consumer.close();
+
+ for (Integer partition : seen.keySet())
+ {
+ Map<String, Integer> byKey = seen.get(partition);
+ for (String key : byKey.keySet())
+ {
+ log.info(
+ "{} - Seen {} messages for partition={}|key={}",
+ id,
+ byKey.get(key),
+ partition,
+ key);
+ }
+ }
+ seen = null;
+
log.info("{} - Consumer-Thread exiting", id);
}
}
- public synchronized void start()
+ public Map<Integer, Map<String, Integer>> getSeen()
+ {
+ return seen;
+ }
+
+ 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
{
- boolean stateChanged = running.compareAndSet(true, false);
- if (!stateChanged)
- throw new RuntimeException("Consumer instance " + id + " is not running!");
-
- log.info("{} - Stopping", id);
- consumer.wakeup();
- future.get();
- log.info("{} - Stopped - consumed {} messages so far", id, consumed);
+ lock.lock();
+ try
+ {
+ if (!running)
+ throw new IllegalStateException("Consumer instance " + id + " is not running!");
+
+ log.info("{} - Stopping", id);
+ consumer.wakeup();
+ condition.await();
+ log.info("{} - Stopped - consumed {} messages so far", id, consumed);
+ }
+ finally
+ {
+ lock.unlock();
+ }
}
@PreDestroy
{
log.info("{} - Was already stopped", id);
}
+ catch (Exception e)
+ {
+ log.error("{} - Unexpected exception while trying to stop the consumer", id, e);
+ }
finally
{
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();
+ }
+ }
}
projects / demos / kafka / training / commitdiff