From: Kai Moritz Date: Sun, 3 Apr 2022 07:23:57 +0000 (+0200) Subject: Setup "entkernt": Implementierungen entfernt X-Git-Tag: docker---lvm-2-tage~4 X-Git-Url: http://juplo.de/gitweb/?a=commitdiff_plain;h=107010572c028dcacf7115b6f90e6afc63be00f8;p=demos%2Fkafka%2Ftraining Setup "entkernt": Implementierungen entfernt --- diff --git a/.gitignore b/.gitignore deleted file mode 100644 index 6240411..0000000 --- a/.gitignore +++ /dev/null @@ -1,3 +0,0 @@ -*.iml -.idea -target diff --git a/README.sh b/README.sh index 0ee50a9..8dcc24c 100755 --- a/README.sh +++ b/README.sh @@ -3,31 +3,12 @@ if [ "$1" = "cleanup" ] then docker-compose down -v - mvn clean exit fi -mvn package || exit 1 -if [ "$1" = "build" ]; then exit; fi - trap 'kill $(jobs -p) 2>/dev/null' EXIT docker-compose up -d echo "Waiting for the Kafka-Cluster to become ready..." docker-compose exec cli cub kafka-ready -b kafka:9092 1 60 > /dev/null 2>&1 || exit 1 - -echo "Producing messages" -mvn exec:java@producer - -echo "Reading messages" -mvn exec:java@consumer & -sleep 7 -kill $(jobs -p) -sleep 2 - -echo "Re-Reading messages" -mvn exec:java@consumer & -sleep 7 -kill $(jobs -p) -sleep 2 diff --git a/pom.xml b/pom.xml deleted file mode 100644 index c104ca1..0000000 --- a/pom.xml +++ /dev/null @@ -1,58 +0,0 @@ - - - - 4.0.0 - - - org.springframework.boot - spring-boot-starter-parent - 2.6.5 - - - - de.juplo.kafka - first-contact - First Contact: a Simple Producer and a simple Consumer-Group - 1.0-SNAPSHOT - - - - org.apache.kafka - kafka-clients - - - org.projectlombok - lombok - - - ch.qos.logback - logback-classic - - - - - - - org.codehaus.mojo - exec-maven-plugin - 3.0.0 - - - producer - - de.juplo.kafka.SimpleProducer - - - - consumer - - de.juplo.kafka.SimpleConsumer - - - - - - - - diff --git a/src/main/java/de/juplo/kafka/SimpleConsumer.java b/src/main/java/de/juplo/kafka/SimpleConsumer.java deleted file mode 100644 index e4d9697..0000000 --- a/src/main/java/de/juplo/kafka/SimpleConsumer.java +++ /dev/null @@ -1,131 +0,0 @@ -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.errors.WakeupException; -import org.apache.kafka.common.serialization.StringDeserializer; - -import java.time.Duration; -import java.util.Arrays; -import java.util.Properties; -import java.util.concurrent.locks.Condition; -import java.util.concurrent.locks.Lock; -import java.util.concurrent.locks.ReentrantLock; - - -@Slf4j -public class SimpleConsumer -{ - private long consumed = 0; - private KafkaConsumer consumer; - private Lock lock = new ReentrantLock(); - private Condition stopped = lock.newCondition(); - - - public SimpleConsumer() - { - // tag::create[] - Properties props = new Properties(); - props.put("bootstrap.servers", ":9092"); - props.put("group.id", "my-consumer"); // << Used for Offset-Commits - // end::create[] - props.put("auto.offset.reset", "earliest"); - // tag::create[] - props.put("key.deserializer", StringDeserializer.class.getName()); - props.put("value.deserializer", StringDeserializer.class.getName()); - - KafkaConsumer consumer = new KafkaConsumer<>(props); - // end::create[] - this.consumer = consumer; - } - - - public void run() - { - String id = "C"; - - try - { - log.info("{} - Subscribing to topic test", id); - consumer.subscribe(Arrays.asList("test")); - - // tag::loop[] - while (true) - { - ConsumerRecords records = - consumer.poll(Duration.ofSeconds(1)); - - // Do something with the data... - // end::loop[] - log.info("{} - Received {} messages", id, records.count()); - for (ConsumerRecord record : records) - { - consumed++; - log.info( - "{} - {}: {}/{} - {}={}", - id, - record.offset(), - record.topic(), - record.partition(), - record.key(), - record.value() - ); - } - // tag::loop[] - } - // end::loop[] - } - catch(WakeupException e) - { - log.info("{} - RIIING!", id); - } - catch(Exception e) - { - log.error("{} - Unexpected error: {}", id, e.toString()); - } - finally - { - this.lock.lock(); - try - { - log.info("{} - Closing the KafkaConsumer", id); - consumer.close(); - log.info("C - DONE!"); - stopped.signal(); - } - finally - { - this.lock.unlock(); - log.info("{}: Consumed {} messages in total, exiting!", id, consumed); - } - } - } - - - public static void main(String[] args) throws Exception - { - SimpleConsumer instance = new SimpleConsumer(); - - Runtime.getRuntime().addShutdownHook(new Thread(() -> - { - instance.lock.lock(); - try - { - instance.consumer.wakeup(); - instance.stopped.await(); - } - catch (InterruptedException e) - { - log.warn("Interrrupted while waiting for the consumer to stop!", e); - } - finally - { - instance.lock.unlock(); - } - })); - - instance.run(); - } -} diff --git a/src/main/java/de/juplo/kafka/SimpleProducer.java b/src/main/java/de/juplo/kafka/SimpleProducer.java deleted file mode 100644 index 43a7227..0000000 --- a/src/main/java/de/juplo/kafka/SimpleProducer.java +++ /dev/null @@ -1,117 +0,0 @@ -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; -import java.util.concurrent.ExecutorService; -import java.util.concurrent.Future; - - -@Slf4j -public class SimpleProducer -{ - private final String id; - private final String topic; - private final KafkaProducer producer; - - private long produced = 0; - - public SimpleProducer(String clientId, String topic) - { - // tag::create[] - Properties props = new Properties(); - props.put("bootstrap.servers", "localhost:9092"); - props.put("key.serializer", StringSerializer.class.getName()); - props.put("value.serializer", StringSerializer.class.getName()); - - KafkaProducer producer = new KafkaProducer<>(props); - // end::create[] - - this.id = clientId; - this.topic = topic; - this.producer = producer; - } - - public void run() - { - long i = 0; - - try - { - for (; i < 100 ; i++) - { - send(Long.toString(i%10), Long.toString(i)); - } - - log.info("{} - Done", id); - } - finally - { - log.info("{}: Closing the KafkaProducer", id); - producer.close(); - log.info("{}: Produced {} messages in total, exiting!", id, produced); - } - } - - void send(String key, String value) - { - final long time = System.currentTimeMillis(); - - final ProducerRecord 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 - { - SimpleProducer producer = new SimpleProducer("P", "test"); - producer.run(); - } -} diff --git a/src/main/resources/logback.xml b/src/main/resources/logback.xml deleted file mode 100644 index b8e6780..0000000 --- a/src/main/resources/logback.xml +++ /dev/null @@ -1,17 +0,0 @@ - - - - - - %highlight(%-5level) %m%n - - - - - - - - - - -