Spring/spring-kafka
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-56: Change Template to return ListenableFuture

Browse Source 
Resolves #56

Fix a bunch of conflicts after rebasing
This commit is contained in:
Gary Russell
2016-04-01 17:00:58 -04:00
committed by Artem Bilan
parent 0b495cb5ba
commit fac9ec63c9
9 changed files with 274 additions and 255 deletions
Download Patch File Download Diff File Expand all files Collapse all files

142
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaOperations.java
View File

@@ -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 <K> the key type.
* @param <V> the value type.
@@ -34,57 +31,55 @@ import org.springframework.messaging.Message;
*/
public interface KafkaOperations<K, V> {
// 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<RecordMetadata> send(V data);
ListenableFuture<SendResult<K, V>> 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<RecordMetadata> send(K key, V data);
ListenableFuture<SendResult<K, V>> 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<RecordMetadata> send(int partition, K key, V data);
ListenableFuture<SendResult<K, V>> 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<RecordMetadata> send(String topic, V data);
ListenableFuture<SendResult<K, V>> 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<RecordMetadata> send(String topic, K key, V data);
ListenableFuture<SendResult<K, V>> 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<RecordMetadata> send(String topic, int partition, V data);
ListenableFuture<SendResult<K, V>> 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<K, V> {
* @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<RecordMetadata> send(String topic, int partition, K key, V data);
ListenableFuture<SendResult<K, V>> 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<RecordMetadata> 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<SendResult<K, V>> send(Message<?> message);
/**
* Flush the producer.

43
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaProducerException.java Normal file
View File

@@ -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;
}
}

134
spring-kafka/src/main/java/org/springframework/kafka/core/KafkaTemplate.java
View File

@@ -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<K, V> implements KafkaOperations<K, V> {
private MessageConverter messageConverter = new MessagingMessageConverter();
private final boolean autoFlush;
private volatile Producer<K, V> producer;
private volatile String defaultTopic;
private volatile ProducerListener<K, V> producerListener = new LoggingProducerListener<K, V>();
/**
* 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<K, V> 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<K, V> producerFactory, boolean autoFlush) {
this.producerFactory = producerFactory;
this.autoFlush = autoFlush;
}
/**
@@ -109,112 +125,51 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
}
@Override
public Future<RecordMetadata> send(V data) {
public ListenableFuture<SendResult<K, V>> send(V data) {
return send(this.defaultTopic, data);
}
@Override
public Future<RecordMetadata> send(K key, V data) {
public ListenableFuture<SendResult<K, V>> send(K key, V data) {
return send(this.defaultTopic, key, data);
}
@Override
public Future<RecordMetadata> send(int partition, K key, V data) {
public ListenableFuture<SendResult<K, V>> send(int partition, K key, V data) {
return send(this.defaultTopic, partition, key, data);
}
@Override
public Future<RecordMetadata> send(String topic, V data) {
public ListenableFuture<SendResult<K, V>> send(String topic, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, data);
return doSend(producerRecord);
}
@Override
public Future<RecordMetadata> send(String topic, K key, V data) {
public ListenableFuture<SendResult<K, V>> send(String topic, K key, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, key, data);
return doSend(producerRecord);
}
@Override
public Future<RecordMetadata> send(String topic, int partition, V data) {
public ListenableFuture<SendResult<K, V>> send(String topic, int partition, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<K, V>(topic, partition, null, data);
return doSend(producerRecord);
}
@Override
public Future<RecordMetadata> send(String topic, int partition, K key, V data) {
public ListenableFuture<SendResult<K, V>> send(String topic, int partition, K key, V data) {
ProducerRecord<K, V> producerRecord = new ProducerRecord<>(topic, partition, key, data);
return doSend(producerRecord);
}
@SuppressWarnings("unchecked")
@Override
public Future<RecordMetadata> convertAndSend(Message<?> message) {
public ListenableFuture<SendResult<K, V>> send(Message<?> message) {
ProducerRecord<?, ?> producerRecord = this.messageConverter.fromMessage(message, this.defaultTopic);
return doSend((ProducerRecord<K, V>) producerRecord);
}
@Override
public RecordMetadata syncSend(V data) throws InterruptedException, ExecutionException {
Future<RecordMetadata> future = send(data);
flush();
return future.get();
}
@Override
public RecordMetadata syncSend(K key, V data) throws InterruptedException, ExecutionException {
Future<RecordMetadata> future = send(key, data);
flush();
return future.get();
}
@Override
public RecordMetadata syncSend(int partition, K key, V data)
throws InterruptedException, ExecutionException {
Future<RecordMetadata> future = send(partition, key, data);
flush();
return future.get();
}
@Override
public RecordMetadata syncSend(String topic, V data) throws InterruptedException, ExecutionException {
Future<RecordMetadata> future = send(topic, data);
flush();
return future.get();
}
@Override
public RecordMetadata syncSend(String topic, K key, V data)
throws InterruptedException, ExecutionException {
Future<RecordMetadata> future = send(topic, key, data);
flush();
return future.get();
}
@Override
public RecordMetadata syncSend(String topic, int partition, V data)
throws InterruptedException, ExecutionException {
Future<RecordMetadata> 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<RecordMetadata> future = send(topic, partition, key, data);
flush();
return future.get();
}
@Override
public RecordMetadata syncConvertAndSend(Message<?> message)
throws InterruptedException, ExecutionException {
Future<RecordMetadata> future = convertAndSend(message);
flush();
return future.get();
}
@Override
public void flush() {
this.producer.flush();
@@ -225,7 +180,7 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
* @param producerRecord the producer record.
* @return a Future for the {@link RecordMetadata}.
*/
protected Future<RecordMetadata> doSend(ProducerRecord<K, V> producerRecord) {
protected ListenableFuture<SendResult<K, V>> doSend(final ProducerRecord<K, V> producerRecord) {
if (this.producer == null) {
synchronized (this) {
if (this.producer == null) {
@@ -236,14 +191,31 @@ public class KafkaTemplate<K, V> implements KafkaOperations<K, V> {
if (this.logger.isTraceEnabled()) {
this.logger.trace("Sending: " + producerRecord);
}
Future<RecordMetadata> 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<SendResult<K, V>> 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);

6
spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListener.java
View File

@@ -55,4 +55,10 @@ public interface ProducerListener<K, V> {
*/
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();
}

5
spring-kafka/src/main/java/org/springframework/kafka/support/ProducerListenerAdapter.java
View File

@@ -38,4 +38,9 @@ public abstract class ProducerListenerAdapter<K, V> implements ProducerListener<
public void onError(String topic, Integer partition, K key, V value, Exception exception) {
}
@Override
public boolean isInterestedInSuccess() {
return false;
}
}

50
spring-kafka/src/main/java/org/springframework/kafka/support/SendResult.java Normal file
View File

@@ -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 <K> the key type.
* @param <V> the value type.
*
* @author Gary Russell
*
*/
public class SendResult<K, V> {
private final ProducerRecord<K, V> producerRecord;
private final RecordMetadata recordMetadata;
public SendResult(ProducerRecord<K, V> producerRecord, RecordMetadata recordMetadata) {
this.producerRecord = producerRecord;
this.recordMetadata = recordMetadata;
}
public ProducerRecord<K, V> getProducerRecord() {
return this.producerRecord;
}
public RecordMetadata getRecordMetadata() {
return this.recordMetadata;
}
}

2
spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java
View File

@@ -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)

64
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java
View File

@@ -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<Integer, String> consumer;
@BeforeClass
public static void setUp() throws Exception {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testT", "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(
consumerProps);
Consumer<Integer, String> consumer = cf.createConsumer();
consumer = cf.createConsumer();
embeddedKafka.consumeFromAllEmbeddedTopics(consumer);
}
@Test
public void testTemplate() throws Exception {
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
KafkaTemplate<Integer, String> 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<Integer, String> 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<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic(TEMPLATE_TOPIC);
ListenableFuture<SendResult<Integer, String>> future = template.send("foo");
template.flush();
final CountDownLatch latch = new CountDownLatch(1);
final AtomicReference<SendResult<Integer, String>> theResult = new AtomicReference<>();
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
@Override
public void onSuccess(SendResult<Integer, String> 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();
}
}

83
src/reference/asciidoc/kafka.adoc
View File

@@ -8,50 +8,21 @@ Both asynchronous and synchronous methods are provided, with the async methods r
[source, java]
----
// Async methods
ListenableFuture<SendResult<K, V>> send(V data);
Future<RecordMetadata> send(V data);
ListenableFuture<SendResult<K, V>> send(K key, V data);
Future<RecordMetadata> send(K key, V data);
ListenableFuture<SendResult<K, V>> send(int partition, K key, V data);
Future<RecordMetadata> send(int partition, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, V data);
Future<RecordMetadata> send(String topic, V data);
ListenableFuture<SendResult<K, V>> send(String topic, K key, V data);
Future<RecordMetadata> send(String topic, K key, V data);
ListenableFuture<SendResult<K, V>> send(String topic, int partition, V data);
Future<RecordMetadata> send(String topic, int partition, V data);
ListenableFuture<SendResult<K, V>> send(String topic, int partition, K key, V data);
Future<RecordMetadata> send(String topic, int partition, K key, V data);
Future<RecordMetadata> 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<SendResult<K, V>> 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<K, V> {
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<K, V> {
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<SendResult>`.
You can register a callback with the listener to receive the result of the send asynchronously.
[source, java]
----
ListenableFuture<SendResult<Integer, String>> future = template.send("foo");
future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {
@Override
public void onSuccess(SendResult<Integer, String> 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