X-Git-Url: https://juplo.de/gitweb/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fde%2Fjuplo%2Fkafka%2FEndlessProducer.java;h=9c69f31e2688688319a5757e10b9bba9b7b8c9d9;hb=refs%2Fheads%2Fendless-long-producer;hp=cc3150e95ccb5555f1b8baa1dd1d702f8894cdc3;hpb=ca946f746344bdff7d92325d56a013729b6055a3;p=demos%2Fkafka%2Ftraining diff --git a/src/main/java/de/juplo/kafka/EndlessProducer.java b/src/main/java/de/juplo/kafka/EndlessProducer.java index cc3150e..9c69f31 100644 --- a/src/main/java/de/juplo/kafka/EndlessProducer.java +++ b/src/main/java/de/juplo/kafka/EndlessProducer.java @@ -3,13 +3,13 @@ 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.LongSerializer; import org.apache.kafka.common.serialization.StringSerializer; import javax.annotation.PreDestroy; import java.util.Properties; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; -import java.util.concurrent.Future; @Slf4j @@ -19,12 +19,11 @@ public class EndlessProducer implements Runnable private final String id; private final String topic; private final int throttleMs; - private final KafkaProducer producer; + private final KafkaProducer producer; private boolean running = false; private long i = 0; private long produced = 0; - private Future future = null; public EndlessProducer( ExecutorService executor, @@ -44,7 +43,7 @@ public class EndlessProducer implements Runnable props.put("client.id", clientId); props.put("acks", acks); props.put("key.serializer", StringSerializer.class.getName()); - props.put("value.serializer", StringSerializer.class.getName()); + props.put("value.serializer", LongSerializer.class.getName()); this.producer = new KafkaProducer<>(props); } @@ -56,7 +55,7 @@ public class EndlessProducer implements Runnable { for (; running; i++) { - send(Long.toString(i%10), Long.toString(i)); + send(Long.toString(i%10), i); if (throttleMs > 0) { @@ -75,16 +74,24 @@ public class EndlessProducer implements Runnable } catch (Exception e) { - + log.error("{} - Unexpected Exception:", id, e); + } + finally + { + synchronized (this) + { + running = false; + log.info("{} - Stopped - produced {} messages so far", id, produced); + } } } - void send(String key, String value) + void send(String key, Long value) { final long time = System.currentTimeMillis(); - final ProducerRecord record = new ProducerRecord<>( - "test", // Topic + final ProducerRecord record = new ProducerRecord<>( + topic, // Topic key, // Key value // Value ); @@ -138,7 +145,7 @@ public class EndlessProducer implements Runnable log.info("{} - Starting - produced {} messages before", id, produced); running = true; - future = executor.submit(this); + executor.submit(this); } public synchronized void stop() throws ExecutionException, InterruptedException @@ -148,8 +155,6 @@ public class EndlessProducer implements Runnable log.info("{} - Stopping...", id); running = false; - future.get(); - log.info("{} - Stopped - produced {} messages so far", id, produced); } @PreDestroy