From fac9ec63c95d553bf9814ef7412a965c5a43ca4e Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Fri, 1 Apr 2016 17:00:58 -0400 Subject: [PATCH] GH-56: Change Template to return ListenableFuture Resolves #56 Fix a bunch of conflicts after rebasing --- .../kafka/core/KafkaOperations.java | 142 +++--------------- .../kafka/core/KafkaProducerException.java | 43 ++++++ .../kafka/core/KafkaTemplate.java | 134 +++++++---------- .../kafka/support/ProducerListener.java | 6 + .../support/ProducerListenerAdapter.java | 5 + .../kafka/support/SendResult.java | 50 ++++++ .../EnableKafkaIntegrationTests.java | 2 +- .../kafka/core/KafkaTemplateTests.java | 64 ++++++-- src/reference/asciidoc/kafka.adoc | 83 +++++----- 9 files changed, 274 insertions(+), 255 deletions(-) create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/core/KafkaProducerException.java create mode 100644 spring-kafka/src/main/java/org/springframework/kafka/support/SendResult.java diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java index 935a24d8..1efd1524 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java @@ -16,15 +16,12 @@ package org.springframework.kafka.core; -import java.util.concurrent.ExecutionException; -import java.util.concurrent.Future; - -import org.apache.kafka.clients.producer.RecordMetadata; - +import org.springframework.kafka.support.SendResult; import org.springframework.messaging.Message; +import org.springframework.util.concurrent.ListenableFuture; /** - * The basic Kafka operations contract. + * The basic Kafka operations contract returning {@link ListenableFuture}s. * * @param the key type. * @param the value type. @@ -34,57 +31,55 @@ import org.springframework.messaging.Message; */ public interface KafkaOperations { - // Async methods - /** * Send the data to the default topic with no key or partition. * @param data The data. - * @return a Future for the {@link RecordMetadata}. + * @return a Future for the {@link SendResult}. */ - Future send(V data); + ListenableFuture> send(V data); /** * Send the data to the default topic with the provided key and no partition. * @param key the key. * @param data The data. - * @return a Future for the {@link RecordMetadata}. + * @return a Future for the {@link SendResult}. */ - Future send(K key, V data); + ListenableFuture> send(K key, V data); /** * Send the data to the default topic with the provided key and partition. * @param partition the partition. * @param key the key. * @param data the data. - * @return a Future for the {@link RecordMetadata}. + * @return a Future for the {@link SendResult}. */ - Future send(int partition, K key, V data); + ListenableFuture> send(int partition, K key, V data); /** * Send the data to the provided topic with no key or partition. * @param topic the topic. * @param data The data. - * @return a Future for the {@link RecordMetadata}. + * @return a Future for the {@link SendResult}. */ - Future send(String topic, V data); + ListenableFuture> send(String topic, V data); /** * Send the data to the provided topic with the provided key and no partition. * @param topic the topic. * @param key the key. * @param data The data. - * @return a Future for the {@link RecordMetadata}. + * @return a Future for the {@link SendResult}. */ - Future send(String topic, K key, V data); + ListenableFuture> send(String topic, K key, V data); /** * Send the data to the provided topic with the provided partition and no key. * @param topic the topic. * @param partition the partition. * @param data The data. - * @return a Future for the {@link RecordMetadata}. + * @return a Future for the {@link SendResult}. */ - Future send(String topic, int partition, V data); + ListenableFuture> send(String topic, int partition, V data); /** * Send the data to the provided topic with the provided key and partition. @@ -92,119 +87,20 @@ public interface KafkaOperations { * @param partition the partition. * @param key the key. * @param data the data. - * @return a Future for the {@link RecordMetadata}. + * @return a Future for the {@link SendResult}. */ - Future send(String topic, int partition, K key, V data); + ListenableFuture> send(String topic, int partition, K key, V data); /** * Send a message with routing information in message headers. The message payload * may be converted before sending. * @param message the message to send. - * @return a Future for the {@link RecordMetadata}. + * @return a Future for the {@link SendResult}. * @see org.springframework.kafka.support.KafkaHeaders#TOPIC * @see org.springframework.kafka.support.KafkaHeaders#PARTITION_ID * @see org.springframework.kafka.support.KafkaHeaders#MESSAGE_KEY */ - Future convertAndSend(Message message); - - - // Sync methods - - /** - * Send the data to the default topic with no key or partition; - * wait for result. - * @param data The data. - * @return a {@link RecordMetadata}. - * @throws ExecutionException execution exception while awaiting result. - * @throws InterruptedException thread interrupted while awaiting result. - */ - RecordMetadata syncSend(V data) throws InterruptedException, ExecutionException; - - /** - * Send the data to the default topic with the provided key and no partition; - * wait for result. - * @param key the key. - * @param data The data. - * @return a {@link RecordMetadata}. - * @throws ExecutionException execution exception while awaiting result. - * @throws InterruptedException thread interrupted while awaiting result. - */ - RecordMetadata syncSend(K key, V data) throws InterruptedException, ExecutionException; - - /** - * Send the data to the default topic with the provided key and partition. - * wait for result. - * @param partition the partition. - * @param key the key. - * @param data the data. - * @return a {@link RecordMetadata}. - * @throws ExecutionException execution exception while awaiting result. - * @throws InterruptedException thread interrupted while awaiting result. - */ - RecordMetadata syncSend(int partition, K key, V data) throws InterruptedException, ExecutionException; - - /** - * Send the data to the provided topic with no key or partition; - * wait for result. - * @param topic the topic. - * @param data The data. - * @return a {@link RecordMetadata}. - * @throws ExecutionException execution exception while awaiting result. - * @throws InterruptedException thread interrupted while awaiting result. - */ - RecordMetadata syncSend(String topic, V data) throws InterruptedException, ExecutionException; - - /** - * Send the data to the provided topic with the provided key and no partition; - * wait for result. - * @param topic the topic. - * @param key the key. - * @param data The data. - * @return a {@link RecordMetadata}. - * @throws ExecutionException execution exception while awaiting result. - * @throws InterruptedException thread interrupted while awaiting result. - */ - RecordMetadata syncSend(String topic, K key, V data) throws InterruptedException, ExecutionException; - - /** - * Send the data to the provided topic with the provided partition and no key; - * wait for result. - * @param topic the topic. - * @param partition the partition. - * @param data The data. - * @return a {@link RecordMetadata}. - * @throws ExecutionException execution exception while awaiting result. - * @throws InterruptedException thread interrupted while awaiting result. - */ - RecordMetadata syncSend(String topic, int partition, V data) throws InterruptedException, ExecutionException; - - /** - * Send the data to the provided topic with the provided key and partition; - * wait for result. - * @param topic the topic. - * @param partition the partition. - * @param key the key. - * @param data the data. - * @return a {@link RecordMetadata}. - * @throws ExecutionException execution exception while awaiting result. - * @throws InterruptedException thread interrupted while awaiting result. - */ - RecordMetadata syncSend(String topic, int partition, K key, V data) - throws InterruptedException, ExecutionException; - - /** - * Send a message with routing information in message headers. The message payload - * may be converted before sending. - * @param message the message to send. - * @return a Future for the {@link RecordMetadata}. - * @throws ExecutionException execution exception while awaiting result. - * @throws InterruptedException thread interrupted while awaiting result. - * @see org.springframework.kafka.support.KafkaHeaders#TOPIC - * @see org.springframework.kafka.support.KafkaHeaders#PARTITION_ID - * @see org.springframework.kafka.support.KafkaHeaders#MESSAGE_KEY - */ - RecordMetadata syncConvertAndSend(Message message) - throws InterruptedException, ExecutionException; + ListenableFuture> send(Message message); /** * Flush the producer. diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaProducerException.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaProducerException.java new file mode 100644 index 00000000..b2c35ebe --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaProducerException.java @@ -0,0 +1,43 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.core; + +import org.apache.kafka.clients.producer.ProducerRecord; + +import org.springframework.kafka.KafkaException; + +/** + * Exceptions when producing. + * + * @author Gary Russell + * + */ +@SuppressWarnings("serial") +public class KafkaProducerException extends KafkaException { + + private final ProducerRecord producerRecord; + + public KafkaProducerException(ProducerRecord failedProducerRecord, String message, Throwable cause) { + super(message, cause); + this.producerRecord = failedProducerRecord; + } + + public ProducerRecord getProducerRecord() { + return this.producerRecord; + } + +} diff --git a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java index 61a86183..5a6796ce 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java @@ -16,21 +16,23 @@ package org.springframework.kafka.core; -import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; +import org.apache.kafka.clients.producer.Callback; import org.apache.kafka.clients.producer.Producer; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.clients.producer.RecordMetadata; import org.springframework.kafka.support.LoggingProducerListener; import org.springframework.kafka.support.ProducerListener; -import org.springframework.kafka.support.ProducerListenerInvokingCallback; +import org.springframework.kafka.support.SendResult; import org.springframework.kafka.support.converter.MessageConverter; import org.springframework.kafka.support.converter.MessagingMessageConverter; import org.springframework.messaging.Message; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.SettableListenableFuture; /** @@ -50,18 +52,32 @@ public class KafkaTemplate implements KafkaOperations { private MessageConverter messageConverter = new MessagingMessageConverter(); +private final boolean autoFlush; private volatile Producer producer; private volatile String defaultTopic; private volatile ProducerListener producerListener = new LoggingProducerListener(); + /** - * Create an instance using the supplied producer factory. + * Create an instance using the supplied producer factory and autoFlush false. * @param producerFactory the producer factory. */ public KafkaTemplate(ProducerFactory producerFactory) { + this(producerFactory, false); + } + + /** + * Create an instance using the supplied producer factory and autoFlush setting. + * Set autoFlush to true if you wish to synchronously interact with Kafaka, calling + * {@link Future#get()} on the result. + * @param producerFactory the producer factory. + * @param autoFlush true to flush after each send. + */ + public KafkaTemplate(ProducerFactory producerFactory, boolean autoFlush) { this.producerFactory = producerFactory; + this.autoFlush = autoFlush; } /** @@ -109,112 +125,51 @@ public class KafkaTemplate implements KafkaOperations { } @Override - public Future send(V data) { + public ListenableFuture> send(V data) { return send(this.defaultTopic, data); } @Override - public Future send(K key, V data) { + public ListenableFuture> send(K key, V data) { return send(this.defaultTopic, key, data); } @Override - public Future send(int partition, K key, V data) { + public ListenableFuture> send(int partition, K key, V data) { return send(this.defaultTopic, partition, key, data); } @Override - public Future send(String topic, V data) { + public ListenableFuture> send(String topic, V data) { ProducerRecord producerRecord = new ProducerRecord<>(topic, data); return doSend(producerRecord); } @Override - public Future send(String topic, K key, V data) { + public ListenableFuture> send(String topic, K key, V data) { ProducerRecord producerRecord = new ProducerRecord<>(topic, key, data); return doSend(producerRecord); } @Override - public Future send(String topic, int partition, V data) { + public ListenableFuture> send(String topic, int partition, V data) { ProducerRecord producerRecord = new ProducerRecord(topic, partition, null, data); return doSend(producerRecord); } @Override - public Future send(String topic, int partition, K key, V data) { + public ListenableFuture> send(String topic, int partition, K key, V data) { ProducerRecord producerRecord = new ProducerRecord<>(topic, partition, key, data); return doSend(producerRecord); } @SuppressWarnings("unchecked") @Override - public Future convertAndSend(Message message) { + public ListenableFuture> send(Message message) { ProducerRecord producerRecord = this.messageConverter.fromMessage(message, this.defaultTopic); return doSend((ProducerRecord) producerRecord); } - @Override - public RecordMetadata syncSend(V data) throws InterruptedException, ExecutionException { - Future future = send(data); - flush(); - return future.get(); - } - - @Override - public RecordMetadata syncSend(K key, V data) throws InterruptedException, ExecutionException { - Future future = send(key, data); - flush(); - return future.get(); - } - - @Override - public RecordMetadata syncSend(int partition, K key, V data) - throws InterruptedException, ExecutionException { - Future future = send(partition, key, data); - flush(); - return future.get(); - } - - @Override - public RecordMetadata syncSend(String topic, V data) throws InterruptedException, ExecutionException { - Future future = send(topic, data); - flush(); - return future.get(); - } - - @Override - public RecordMetadata syncSend(String topic, K key, V data) - throws InterruptedException, ExecutionException { - Future future = send(topic, key, data); - flush(); - return future.get(); - } - - @Override - public RecordMetadata syncSend(String topic, int partition, V data) - throws InterruptedException, ExecutionException { - Future future = send(topic, partition, data); - flush(); - return future.get(); - } - - @Override - public RecordMetadata syncSend(String topic, int partition, K key, V data) - throws InterruptedException, ExecutionException { - Future future = send(topic, partition, key, data); - flush(); - return future.get(); - } - - @Override - public RecordMetadata syncConvertAndSend(Message message) - throws InterruptedException, ExecutionException { - Future future = convertAndSend(message); - flush(); - return future.get(); - } - @Override public void flush() { this.producer.flush(); @@ -225,7 +180,7 @@ public class KafkaTemplate implements KafkaOperations { * @param producerRecord the producer record. * @return a Future for the {@link RecordMetadata}. */ - protected Future doSend(ProducerRecord producerRecord) { + protected ListenableFuture> doSend(final ProducerRecord producerRecord) { if (this.producer == null) { synchronized (this) { if (this.producer == null) { @@ -236,14 +191,31 @@ public class KafkaTemplate implements KafkaOperations { if (this.logger.isTraceEnabled()) { this.logger.trace("Sending: " + producerRecord); } - Future future; - if (this.producerListener == null) { - future = this.producer.send(producerRecord); - } - else { - future = this.producer.send(producerRecord, - new ProducerListenerInvokingCallback<>(producerRecord.topic(), producerRecord.partition(), - producerRecord.key(), producerRecord.value(), this.producerListener)); + final SettableListenableFuture> future = new SettableListenableFuture<>(); + this.producer.send(producerRecord, new Callback() { + + @Override + public void onCompletion(RecordMetadata metadata, Exception exception) { + if (exception == null) { + future.set(new SendResult<>(producerRecord, metadata)); + if (KafkaTemplate.this.producerListener != null + && KafkaTemplate.this.producerListener.isInterestedInSuccess()) { + KafkaTemplate.this.producerListener.onSuccess(producerRecord.topic(), + producerRecord.partition(), producerRecord.key(), producerRecord.value(), metadata); + } + } + else { + future.setException(new KafkaProducerException(producerRecord, "Failed to send", exception)); + if (KafkaTemplate.this.producerListener != null) { + KafkaTemplate.this.producerListener.onError(producerRecord.topic(), + producerRecord.partition(), producerRecord.key(), producerRecord.value(), exception); + } + } + } + + }); + if (this.autoFlush) { + flush(); } if (this.logger.isTraceEnabled()) { this.logger.trace("Sent: " + producerRecord); diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java index 7b8e3f8a..8dba95cb 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java @@ -55,4 +55,10 @@ public interface ProducerListener { */ void onError(String topic, Integer partition, K key, V value, Exception exception); + /** + * Return true if this listener is interested in success as well as failure. + * @return true to express interest in successful sends. + */ + boolean isInterestedInSuccess(); + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerAdapter.java b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerAdapter.java index 6b2377bd..6fbbbbc2 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerAdapter.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerAdapter.java @@ -38,4 +38,9 @@ public abstract class ProducerListenerAdapter implements ProducerListener< public void onError(String topic, Integer partition, K key, V value, Exception exception) { } + @Override + public boolean isInterestedInSuccess() { + return false; + } + } diff --git a/spring-kafka/src/main/java/org/springframework/kafka/support/SendResult.java b/spring-kafka/src/main/java/org/springframework/kafka/support/SendResult.java new file mode 100644 index 00000000..443d30ca --- /dev/null +++ b/spring-kafka/src/main/java/org/springframework/kafka/support/SendResult.java @@ -0,0 +1,50 @@ +/* + * Copyright 2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.kafka.support; + +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.kafka.clients.producer.RecordMetadata; + +/** + * Result for a Listenablefuture after a send. + * + * @param the key type. + * @param the value type. + * + * @author Gary Russell + * + */ +public class SendResult { + + private final ProducerRecord producerRecord; + + private final RecordMetadata recordMetadata; + + public SendResult(ProducerRecord producerRecord, RecordMetadata recordMetadata) { + this.producerRecord = producerRecord; + this.recordMetadata = recordMetadata; + } + + public ProducerRecord getProducerRecord() { + return this.producerRecord; + } + + public RecordMetadata getRecordMetadata() { + return this.recordMetadata; + } + +} diff --git a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java index b2b1c13d..1f0b8bf7 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java @@ -157,7 +157,7 @@ public class EnableKafkaIntegrationTests { public void testJson() throws Exception { Foo foo = new Foo(); foo.setBar("bar"); - kafkaJsonTemplate.convertAndSend(MessageBuilder.withPayload(foo) + kafkaJsonTemplate.send(MessageBuilder.withPayload(foo) .setHeader(KafkaHeaders.TOPIC, "annotated10") .setHeader(KafkaHeaders.PARTITION_ID, 0) .setHeader(KafkaHeaders.MESSAGE_KEY, 2) diff --git a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java index 46eac982..3909ef2c 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java @@ -24,18 +24,23 @@ import static org.springframework.kafka.test.assertj.KafkaConditions.value; import java.util.Map; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import org.apache.kafka.clients.consumer.Consumer; import org.apache.kafka.clients.consumer.ConsumerRecord; import org.apache.kafka.clients.producer.RecordMetadata; +import org.junit.BeforeClass; import org.junit.ClassRule; import org.junit.Test; import org.springframework.kafka.support.KafkaHeaders; import org.springframework.kafka.support.ProducerListenerAdapter; +import org.springframework.kafka.support.SendResult; import org.springframework.kafka.test.rule.KafkaEmbedded; import org.springframework.kafka.test.utils.KafkaTestUtils; import org.springframework.messaging.support.MessageBuilder; +import org.springframework.util.concurrent.ListenableFuture; +import org.springframework.util.concurrent.ListenableFutureCallback; /** @@ -50,36 +55,42 @@ public class KafkaTemplateTests { @ClassRule public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, TEMPLATE_TOPIC); - @Test - public void testTemplate() throws Exception { + private static Consumer consumer; + + @BeforeClass + public static void setUp() throws Exception { Map consumerProps = KafkaTestUtils.consumerProps("testT", "false", embeddedKafka); DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory( consumerProps); - Consumer consumer = cf.createConsumer(); + consumer = cf.createConsumer(); embeddedKafka.consumeFromAllEmbeddedTopics(consumer); + } + + @Test + public void testTemplate() throws Exception { Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); - KafkaTemplate template = new KafkaTemplate<>(pf); + KafkaTemplate template = new KafkaTemplate<>(pf, true); template.setDefaultTopic(TEMPLATE_TOPIC); - template.syncSend("foo"); + template.send("foo"); assertThat(KafkaTestUtils.getSingleRecord(consumer, TEMPLATE_TOPIC)).has(value("foo")); - template.syncSend(0, 2, "bar"); + template.send(0, 2, "bar"); ConsumerRecord received = KafkaTestUtils.getSingleRecord(consumer, TEMPLATE_TOPIC); assertThat(received).has(key(2)); assertThat(received).has(partition(0)); assertThat(received).has(value("bar")); - template.syncSend(TEMPLATE_TOPIC, 0, 2, "baz"); + template.send(TEMPLATE_TOPIC, 0, 2, "baz"); received = KafkaTestUtils.getSingleRecord(consumer, TEMPLATE_TOPIC); assertThat(received).has(key(2)); assertThat(received).has(partition(0)); assertThat(received).has(value("baz")); - template.syncSend(TEMPLATE_TOPIC, 0, "qux"); + template.send(TEMPLATE_TOPIC, 0, "qux"); received = KafkaTestUtils.getSingleRecord(consumer, TEMPLATE_TOPIC); assertThat(received).has(key((Integer) null)); assertThat(received).has(partition(0)); assertThat(received).has(value("qux")); - template.syncConvertAndSend(MessageBuilder.withPayload("fiz") + template.send(MessageBuilder.withPayload("fiz") .setHeader(KafkaHeaders.TOPIC, TEMPLATE_TOPIC) .setHeader(KafkaHeaders.PARTITION_ID, 0) .setHeader(KafkaHeaders.MESSAGE_KEY, 2) @@ -88,7 +99,7 @@ public class KafkaTemplateTests { assertThat(received).has(key(2)); assertThat(received).has(partition(0)); assertThat(received).has(value("fiz")); - template.syncConvertAndSend(MessageBuilder.withPayload("buz") + template.send(MessageBuilder.withPayload("buz") .setHeader(KafkaHeaders.PARTITION_ID, 0) .setHeader(KafkaHeaders.MESSAGE_KEY, 2) .build()); @@ -115,10 +126,41 @@ public class KafkaTemplateTests { latch.countDown(); } + @Override + public boolean isInterestedInSuccess() { + return true; + } + }); - template.syncSend("foo"); + template.send("foo"); template.flush(); assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); } + @Test + public void testWithCallback() throws Exception { + Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); + ProducerFactory pf = new DefaultKafkaProducerFactory(senderProps); + KafkaTemplate template = new KafkaTemplate<>(pf, true); + template.setDefaultTopic(TEMPLATE_TOPIC); + ListenableFuture> future = template.send("foo"); + template.flush(); + final CountDownLatch latch = new CountDownLatch(1); + final AtomicReference> theResult = new AtomicReference<>(); + future.addCallback(new ListenableFutureCallback>() { + + @Override + public void onSuccess(SendResult result) { + theResult.set(result); + latch.countDown(); + } + + @Override + public void onFailure(Throwable ex) { + } + + }); + assertThat(KafkaTestUtils.getSingleRecord(consumer, TEMPLATE_TOPIC)).has(value("foo")); + assertThat(latch.await(10, TimeUnit.SECONDS)).isTrue(); + } } diff --git a/src/reference/asciidoc/kafka.adoc b/src/reference/asciidoc/kafka.adoc index 752a58b2..84b444c8 100644 --- a/src/reference/asciidoc/kafka.adoc +++ b/src/reference/asciidoc/kafka.adoc @@ -8,50 +8,21 @@ Both asynchronous and synchronous methods are provided, with the async methods r [source, java] ---- -// Async methods +ListenableFuture> send(V data); -Future send(V data); +ListenableFuture> send(K key, V data); -Future send(K key, V data); +ListenableFuture> send(int partition, K key, V data); -Future send(int partition, K key, V data); +ListenableFuture> send(String topic, V data); -Future send(String topic, V data); +ListenableFuture> send(String topic, K key, V data); -Future send(String topic, K key, V data); +ListenableFuture> send(String topic, int partition, V data); -Future send(String topic, int partition, V data); +ListenableFuture> send(String topic, int partition, K key, V data); -Future send(String topic, int partition, K key, V data); - -Future send(Message message); - -// Sync methods - - -RecordMetadata syncSend(V data) - throws InterruptedException, ExecutionException; - -RecordMetadata syncSend(K key, V data) - throws InterruptedException, ExecutionException; - -RecordMetadata syncSend(int partition, K key, V data) - throws InterruptedException, ExecutionException; - -RecordMetadata syncSend(String topic, V data) - throws InterruptedException, ExecutionException; - -RecordMetadata syncSend(String topic, K key, V data) - throws InterruptedException, ExecutionException; - -RecordMetadata syncSend(String topic, int partition, V data) - throws InterruptedException, ExecutionException; - -RecordMetadata syncSend(String topic, int partition, K key, V data) - throws InterruptedException, ExecutionException; - -RecordMetadata syncSend(Message message) - throws InterruptedException, ExecutionException; +ListenableFuture> send(Message message); // Flush the producer. @@ -101,9 +72,11 @@ results of the send (success or failure) instead of waiting for the `Future` to ---- public interface ProducerListener { - void onSuccess(String topic, Integer partition, K key, V value, RecordMetadata recordMetadata); + void onSuccess(String topic, Integer partition, K key, V value, RecordMetadata recordMetadata); - void onError(String topic, Integer partition, K key, V value, Exception exception); + void onError(String topic, Integer partition, K key, V value, Exception exception); + + boolean isInterestedInSuccess(); } ---- @@ -111,8 +84,40 @@ public interface ProducerListener { By default, the template is configured with a `LoggingProducerListener` which logs errors and does nothing when the send is successful. +`onSuccess` is only called if `isInterestedInSuccess` returns `true`. + For convenience, the abstract `ProducerListenerAdapter` is provided in case you only want to implement one of the methods. +It returns `false` for `isInterestedInSuccess`. + +Notice that the send methods return a `ListenableFuture`. +You can register a callback with the listener to receive the result of the send asynchronously. + +[source, java] +---- +ListenableFuture> future = template.send("foo"); +future.addCallback(new ListenableFutureCallback>() { + + @Override + public void onSuccess(SendResult result) { + ... + } + + @Override + public void onFailure(Throwable ex) { + ... + } + +}); +---- + +The `SendResult` has two properties, a `ProducerRecord` and `RecordMetadata`; refer to the Kafka API documentation +for information about those objects. + +If you wish to block the sending thread, to await the result, you can invoke the future's `get()` method. +You may wish to invoke `flush()` before waiting or, for convenience, the template has a constructor with an `autoFlush` +parameter which will cause the template to `flush()` on each send. +Note, however that flushing will likely significantly reduce performance. ==== Receiving Messages