import org.springframework.web.context.request.async.DeferredResult;
import javax.annotation.PreDestroy;
+import java.math.BigInteger;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
{
private final String id;
private final String topic;
+ private final Integer partition;
private final KafkaProducer<String, String> producer;
private long produced = 0;
{
this.id = properties.getClientId();
this.topic = properties.getTopic();
+ this.partition = properties.getPartition();
Properties props = new Properties();
props.put("bootstrap.servers", properties.getBootstrapServer());
final ProducerRecord<String, String> record = new ProducerRecord<>(
topic, // Topic
+ partition, // Partition
key, // Key
value // Value
);
+ record.headers().add("source", id.getBytes());
+ if (correlationId != null)
+ {
+ record.headers().add("id", BigInteger.valueOf(correlationId).toByteArray());
+ }
+
producer.send(record, (metadata, e) ->
{
long now = System.currentTimeMillis();
long now = System.currentTimeMillis();
log.trace(
- "{} - Queued #{} key={} latency={}ms",
+ "{} - Queued message with key={} latency={}ms",
id,
- value,
record.key(),
now - time
);