FROM openjdk:11-jre
VOLUME /tmp
COPY target/*.jar /opt/app.jar
-COPY target/dependency /opt/libs
-ENTRYPOINT [ "java", "-jar", "/opt/app.jar", "kafka:9092", "test" ]
+COPY target/libs /opt/libs
+ENTRYPOINT [ "java", "-jar", "/opt/app.jar", "kafka:9092", "test", "my-group" ]
CMD [ "DCKR" ]
#!/bin/bash
-IMAGE=juplo/simple-producer:1.0-SNAPSHOT
+IMAGE=juplo/simple-consumer:1.0-SNAPSHOT
if [ "$1" = "cleanup" ]
then
docker-compose exec cli kafkacat -b kafka:9092 -t test -q -c 20 -f'topic=%t\tpartition=%p\toffset=%o\tkey=%k\tvalue=%s\n'
-docker-compose stop producer
-docker-compose exec cli kafkacat -b kafka:9092 -t test -q -e -f'topic=%t\tpartition=%p\toffset=%o\tkey=%k\tvalue=%s\n'
-docker-compose logs producer
+docker-compose stop producer consumer-1 consumer-2 consumer-3
+docker-compose logs consumer-1
+docker-compose logs consumer-2
+docker-compose logs consumer-3
producer:
image: juplo/simple-producer:1.0-SNAPSHOT
command: producer
+
+ consumer-1:
+ image: juplo/simple-consumer:1.0-SNAPSHOT
+ command: consumer-1
+
+ consumer-2:
+ image: juplo/simple-consumer:1.0-SNAPSHOT
+ command: consumer-2
+
+ consumer-3:
+ image: juplo/simple-consumer:1.0-SNAPSHOT
+ command: consumer-3
</parent>
<groupId>de.juplo.kafka</groupId>
- <artifactId>simple-producer</artifactId>
- <name>Super Simple Producer</name>
+ <artifactId>simple-consumer</artifactId>
+ <name>Super Simple Consumer-Group</name>
<version>1.0-SNAPSHOT</version>
<dependencies>
<manifest>
<addClasspath>true</addClasspath>
<classpathPrefix>libs/</classpathPrefix>
- <mainClass>de.juplo.kafka.SimpleProducer</mainClass>
+ <mainClass>de.juplo.kafka.SimpleConsumer</mainClass>
</manifest>
</archive>
</configuration>
--- /dev/null
+package de.juplo.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+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.serialization.StringDeserializer;
+
+import java.time.Duration;
+import java.util.Arrays;
+import java.util.Properties;
+
+
+@Slf4j
+public class SimpleConsumer
+{
+ private final String id;
+ private final String topic;
+ private final KafkaConsumer<String, String> consumer;
+
+ private long consumed = 0;
+ private volatile boolean running = true;
+ private volatile boolean done = false;
+
+ public SimpleConsumer(String broker, String topic, String groupId, String clientId)
+ {
+ Properties props = new Properties();
+ props.put("bootstrap.servers", broker);
+ props.put("group.id", groupId); // ID für die Offset-Commits
+ props.put("client.id", clientId); // Nur zur Wiedererkennung
+ props.put("key.deserializer", StringDeserializer.class.getName());
+ props.put("value.deserializer", StringDeserializer.class.getName());
+
+ consumer = new KafkaConsumer<>(props);
+
+ this.topic = topic;
+ this.id = clientId;
+ }
+
+
+ public void run()
+ {
+ try
+ {
+ log.info("{} - Subscribing to topic test", id);
+ consumer.subscribe(Arrays.asList("test"));
+
+ // tag::loop[]
+ while (running)
+ {
+ ConsumerRecords<String, String> records =
+ consumer.poll(Duration.ofSeconds(1));
+
+ // Do something with the data...
+ // end::loop[]
+ log.info("{} - Received {} messages", id, records.count());
+ for (ConsumerRecord<String, String> record : records)
+ {
+ consumed++;
+ log.info(
+ "{} - {}: {}/{} - {}={}",
+ id,
+ record.offset(),
+ record.topic(),
+ record.partition(),
+ record.key(),
+ record.value()
+ );
+ }
+ // tag::loop[]
+ }
+ // end::loop[]
+ }
+ catch(Exception e)
+ {
+ log.error("{} - Unexpected error: {}", id, e.toString());
+ }
+ finally
+ {
+ log.info("{} - Closing the KafkaConsumer", id);
+ consumer.close();
+ log.info("{}: Consumed {} messages in total, exiting!", id, consumed);
+ done = true;
+ }
+ }
+
+
+ public static void main(String[] args) throws Exception
+ {
+ String broker = ":9092";
+ String topic = "test";
+ String groupId = "my-group";
+ String clientId = "DEV";
+
+ switch (args.length)
+ {
+ case 4:
+ clientId = args[3];
+ case 3:
+ groupId = args[2];
+ case 2:
+ topic = args[1];
+ case 1:
+ broker = args[0];
+ }
+
+
+ SimpleConsumer instance = new SimpleConsumer(broker, topic, groupId, clientId);
+
+ Runtime.getRuntime().addShutdownHook(new Thread(() ->
+ {
+ instance.running = false;
+ while (!instance.done)
+ {
+ log.info("Waiting for main-thread...");
+ try
+ {
+ Thread.sleep(1000);
+ }
+ catch (InterruptedException e) {}
+ }
+ log.info("Shutdown completed.");
+ }));
+
+ log.info(
+ "Running SimpleConsumer: broker={}, topic={}, group-id={}, client-id={}",
+ broker,
+ topic,
+ groupId,
+ clientId);
+ instance.run();
+ }
+}
+++ /dev/null
-package de.juplo.kafka;
-
-import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.serialization.StringSerializer;
-
-import java.util.Properties;
-
-
-@Slf4j
-public class SimpleProducer
-{
- private final String id;
- private final String topic;
- private final KafkaProducer<String, String> producer;
-
- private long produced = 0;
- private volatile boolean running = true;
- private volatile boolean done = false;
-
- public SimpleProducer(String broker, String topic, String clientId)
- {
- Properties props = new Properties();
- props.put("bootstrap.servers", broker);
- props.put("client.id", clientId); // Nur zur Wiedererkennung
- props.put("key.serializer", StringSerializer.class.getName());
- props.put("value.serializer", StringSerializer.class.getName());
-
- producer = new KafkaProducer<>(props);
-
- this.topic = topic;
- this.id = clientId;
- }
-
- public void run()
- {
- long i = 0;
-
- try
- {
- for (; running ; i++)
- {
- send(Long.toString(i%10), Long.toString(i));
- Thread.sleep(500);
- }
- }
- catch (InterruptedException e) {}
- finally
- {
- log.info("{}: Closing the KafkaProducer", id);
- producer.close();
- log.info("{}: Produced {} messages in total, exiting!", id, produced);
- done = true;
- }
- }
-
- void send(String key, String value)
- {
- final long time = System.currentTimeMillis();
-
- final ProducerRecord<String, String> record = new ProducerRecord<>(
- topic, // Topic
- key, // Key
- value // Value
- );
-
- producer.send(record, (metadata, e) ->
- {
- long now = System.currentTimeMillis();
- if (e == null)
- {
- // HANDLE SUCCESS
- produced++;
- log.debug(
- "{} - Sent key={} message={} partition={}/{} timestamp={} latency={}ms",
- id,
- record.key(),
- record.value(),
- metadata.partition(),
- metadata.offset(),
- metadata.timestamp(),
- now - time
- );
- }
- else
- {
- // HANDLE ERROR
- log.error(
- "{} - ERROR key={} timestamp={} latency={}ms: {}",
- id,
- record.key(),
- metadata == null ? -1 : metadata.timestamp(),
- now - time,
- e.toString()
- );
- }
- });
-
- long now = System.currentTimeMillis();
- log.trace(
- "{} - Queued #{} key={} latency={}ms",
- id,
- value,
- record.key(),
- now - time
- );
- }
-
-
- public static void main(String[] args) throws Exception
- {
- String broker = ":9092";
- String topic = "test";
- String clientId = "DEV";
-
- switch (args.length)
- {
- case 3:
- clientId = args[2];
- case 2:
- topic = args[1];
- case 1:
- broker = args[0];
- }
-
- SimpleProducer instance = new SimpleProducer(broker, topic, clientId);
-
- Runtime.getRuntime().addShutdownHook(new Thread(() ->
- {
- instance.running = false;
- while (!instance.done)
- {
- log.info("Waiting for main-thread...");
- try
- {
- Thread.sleep(1000);
- }
- catch (InterruptedException e) {}
- }
- log.info("Shutdown completed.");
- }));
-
- log.info(
- "Running SimpleProducer: broker={}, topic={}, client-id={}",
- broker,
- topic,
- clientId);
- instance.run();
- }
-}
projects / demos / kafka / training / commitdiff