+++ /dev/null
-package de.juplo.kafka;
-
-import lombok.extern.slf4j.Slf4j;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.clients.producer.RecordMetadata;
-import org.springframework.kafka.core.KafkaTemplate;
-import org.springframework.kafka.support.SendResult;
-import org.springframework.util.concurrent.ListenableFuture;
-
-import javax.annotation.PreDestroy;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.ExecutorService;
-
-
-@Slf4j
-public class EndlessProducer implements Runnable
-{
- private final ExecutorService executor;
- private final String id;
- private final int throttleMs;
- private final KafkaTemplate<String, String> kafkaTemplate;
-
- private boolean running = false;
- private long i = 0;
- private long produced = 0;
-
- public EndlessProducer(
- ExecutorService executor,
- String clientId,
- int throttleMs,
- KafkaTemplate<String, String> kafkaTemplate)
- {
- this.executor = executor;
- this.id = clientId;
- this.throttleMs = throttleMs;
- this.kafkaTemplate = kafkaTemplate;
- }
-
- @Override
- public void run()
- {
- try
- {
- for (; running; i++)
- {
- send(Long.toString(i%10), Long.toString(i));
-
- if (throttleMs > 0)
- {
- try
- {
- Thread.sleep(throttleMs);
- }
- catch (InterruptedException e)
- {
- log.warn("{} - Interrupted while throttling!", e);
- }
- }
- }
-
- log.info("{} - Done", id);
- }
- 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)
- {
- final long time = System.currentTimeMillis();
-
- ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.sendDefault(key, value);
- listenableFuture.addCallback(
- result ->
- {
- long now = System.currentTimeMillis();
- RecordMetadata metadata = result.getRecordMetadata();
- ProducerRecord<String, String> record = result.getProducerRecord();
-
- // 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
- );
- },
- e ->
- {
- long now = System.currentTimeMillis();
-
- // HANDLE ERROR
- log.error(
- "{} - ERROR key={} latency={}ms: {}",
- id,
- key,
- now - time,
- e.toString()
- );
- });
-
- long now = System.currentTimeMillis();
- log.trace(
- "{} - Queued #{} key={} latency={}ms",
- id,
- value,
- key,
- now - time
- );
- }
-
- public synchronized void start()
- {
- if (running)
- throw new IllegalStateException("Producer instance " + id + " is already running!");
-
- log.info("{} - Starting - produced {} messages before", id, produced);
- running = true;
- executor.submit(this);
- }
-
- public synchronized void stop() throws ExecutionException, InterruptedException
- {
- if (!running)
- throw new IllegalStateException("Producer instance " + id + " is not running!");
-
- log.info("{} - Stopping...", id);
- running = false;
- }
-
- @PreDestroy
- public void destroy() throws ExecutionException, InterruptedException
- {
- log.info("{} - Destroy!", id);
- try
- {
- stop();
- }
- catch (IllegalStateException e)
- {
- log.info("{} - Was already stopped", id);
- }
- finally
- {
- log.info("{}: Produced {} messages in total, exiting!", id, produced);
- }
- }
-}
--- /dev/null
+package de.juplo.kafka;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.clients.producer.RecordMetadata;
+import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.kafka.support.SendResult;
+import org.springframework.util.concurrent.ListenableFuture;
+
+import javax.annotation.PreDestroy;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+
+
+@Slf4j
+public class EndlessProducer implements Runnable
+{
+ private final ExecutorService executor;
+ private final String id;
+ private final int throttleMs;
+ private final KafkaTemplate<String, String> kafkaTemplate;
+
+ private boolean running = false;
+ private long i = 0;
+ private long produced = 0;
+
+ public EndlessProducer(
+ ExecutorService executor,
+ String clientId,
+ int throttleMs,
+ KafkaTemplate<String, String> kafkaTemplate)
+ {
+ this.executor = executor;
+ this.id = clientId;
+ this.throttleMs = throttleMs;
+ this.kafkaTemplate = kafkaTemplate;
+ }
+
+ @Override
+ public void run()
+ {
+ try
+ {
+ for (; running; i++)
+ {
+ send(Long.toString(i%10), Long.toString(i));
+
+ if (throttleMs > 0)
+ {
+ try
+ {
+ Thread.sleep(throttleMs);
+ }
+ catch (InterruptedException e)
+ {
+ log.warn("{} - Interrupted while throttling!", e);
+ }
+ }
+ }
+
+ log.info("{} - Done", id);
+ }
+ 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)
+ {
+ final long time = System.currentTimeMillis();
+
+ ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.sendDefault(key, value);
+ listenableFuture.addCallback(
+ result ->
+ {
+ long now = System.currentTimeMillis();
+ RecordMetadata metadata = result.getRecordMetadata();
+ ProducerRecord<String, String> record = result.getProducerRecord();
+
+ // 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
+ );
+ },
+ e ->
+ {
+ long now = System.currentTimeMillis();
+
+ // HANDLE ERROR
+ log.error(
+ "{} - ERROR key={} latency={}ms: {}",
+ id,
+ key,
+ now - time,
+ e.toString()
+ );
+ });
+
+ long now = System.currentTimeMillis();
+ log.trace(
+ "{} - Queued #{} key={} latency={}ms",
+ id,
+ value,
+ key,
+ now - time
+ );
+ }
+
+ public synchronized void start()
+ {
+ if (running)
+ throw new IllegalStateException("Producer instance " + id + " is already running!");
+
+ log.info("{} - Starting - produced {} messages before", id, produced);
+ running = true;
+ executor.submit(this);
+ }
+
+ public synchronized void stop() throws ExecutionException, InterruptedException
+ {
+ if (!running)
+ throw new IllegalStateException("Producer instance " + id + " is not running!");
+
+ log.info("{} - Stopping...", id);
+ running = false;
+ }
+
+ @PreDestroy
+ public void destroy() throws ExecutionException, InterruptedException
+ {
+ log.info("{} - Destroy!", id);
+ try
+ {
+ stop();
+ }
+ catch (IllegalStateException e)
+ {
+ log.info("{} - Was already stopped", id);
+ }
+ finally
+ {
+ log.info("{}: Produced {} messages in total, exiting!", id, produced);
+ }
+ }
+}