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

GH-212: Add ConsumerSeekAware impl to Inbounds (#213)

Browse Source 
* GH-212: Add ConsumerSeekAware impl to Inbounds

Fixes spring-projects/spring-integration-kafka#212

**Cherry-pick to 3.0.x**

* GH-212: Add ConsumerSeekAware impl to Inbounds

Fixes spring-projects/spring-integration-kafka#212

* Introduce a new `IntegrationKafkaHeaders.CONSUMER_SEEK_CALLBACK`
header to be populated to messages for sending to the channel
* Populate that header from the `KafkaInboundGateway` and
`KafkaMessageDrivenChannelAdapter` into the message from the
`seekCallBack` property if `ListenerContainer` is single-threaded or
from the `ThreadLocal<ConsumerSeekAware.ConsumerSeekCallback>` otherwise;
and only if newly introduced `setAdditionalHeaders` is `true`
* Populate `seekCallBack` property or `ThreadLocal<?>` from the
`registerSeekCallback()` implementation from the internal listeners
* Add `setOnPartitionsAssignedSeekCallback(BiConsumer)` and
`setOnIdleSeekCallback(BiConsumer)` options to react for the appropriate
event from the underlying container and perform appropriate seek management
* Add new options to the DSL classes and cover them with tests, including
check for new `IntegrationKafkaHeaders.CONSUMER_SEEK_CALLBACK` header

**Cherry-pick to 3.0.x**

* Address PR comments: remove unnecessary API

* *Polishing `setOnPartitionsAssignedSeekCallback()` JavaDocs
*Close producers in the `KafkaProducerMessageHandlerTests`
This commit is contained in:
Artem Bilan
2018-09-07 14:31:59 -04:00
committed by Artem Bilan
parent bedc0c8fdc
commit f29bd3e322
8 changed files with 171 additions and 42 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaInboundGatewaySpec.java
View File

@@ -18,8 +18,11 @@ package org.springframework.integration.kafka.dsl;
import java.util.Collections;
import java.util.Map;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.springframework.integration.dsl.ComponentsRegistration;
import org.springframework.integration.dsl.MessagingGatewaySpec;
import org.springframework.integration.kafka.inbound.KafkaInboundGateway;
@@ -27,6 +30,7 @@ import org.springframework.integration.support.ObjectStringMapBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.kafka.support.converter.RecordMessageConverter;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.support.RetryTemplate;
@@ -41,6 +45,7 @@ import org.springframework.util.Assert;
* @param <S> the target {@link KafkaInboundGatewaySpec} implementation type.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 3.0.2
*/
@@ -91,6 +96,20 @@ public class KafkaInboundGatewaySpec<K, V, R, S extends KafkaInboundGatewaySpec<
return _this();
}
/**
* Specify a {@link BiConsumer} for seeks management during
* {@link ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback)}
* call from the {@link org.springframework.kafka.listener.KafkaMessageListenerContainer}.
* @param onPartitionsAssignedCallback the {@link BiConsumer} to use
* @return the spec
* @since 3.0.4
*/
public S onPartitionsAssignedSeekCallback(
BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) {
this.target.setOnPartitionsAssignedSeekCallback(onPartitionsAssignedCallback);
return _this();
}
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.container, getId() == null ? null : getId() + ".container");

18
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/dsl/KafkaMessageDrivenChannelAdapterSpec.java
View File

@@ -18,13 +18,17 @@ package org.springframework.integration.kafka.dsl;
import java.util.Collections;
import java.util.Map;
import java.util.function.BiConsumer;
import java.util.function.Consumer;
import org.apache.kafka.common.TopicPartition;
import org.springframework.integration.dsl.ComponentsRegistration;
import org.springframework.integration.dsl.MessageProducerSpec;
import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.kafka.listener.adapter.RecordFilterStrategy;
import org.springframework.kafka.support.converter.BatchMessageConverter;
import org.springframework.kafka.support.converter.MessageConverter;
@@ -156,6 +160,20 @@ public class KafkaMessageDrivenChannelAdapterSpec<K, V, S extends KafkaMessageDr
return _this();
}
/**
* Specify a {@link BiConsumer} for seeks management during
* {@link ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback)}
* call from the {@link org.springframework.kafka.listener.KafkaMessageListenerContainer}.
* @param onPartitionsAssignedCallback the {@link BiConsumer} to use
* @return the spec
* @since 3.0.4
*/
public S onPartitionsAssignedSeekCallback(
BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) {
this.target.setOnPartitionsAssignedSeekCallback(onPartitionsAssignedCallback);
return _this();
}
@Override
public Map<Object, String> getComponentsToRegister() {
return Collections.singletonMap(this.container, getId() == null ? null : getId() + ".container");

31
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaInboundGateway.java
View File

@@ -16,10 +16,13 @@
package org.springframework.integration.kafka.inbound;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -32,6 +35,7 @@ import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.adapter.RecordMessagingMessageListenerAdapter;
import org.springframework.kafka.listener.adapter.RetryingMessageListenerAdapter;
@@ -58,6 +62,7 @@ import org.springframework.util.Assert;
* @param <R> the reply value type.
*
* @author Gary Russell
* @author Artem Bilan
*
* @since 3.0.2
*
@@ -76,6 +81,8 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
private RecoveryCallback<? extends Object> recoveryCallback;
private BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedSeekCallback;
/**
* Construct an instance with the provided container.
* @param messageListenerContainer the container.
@@ -133,6 +140,21 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
this.recoveryCallback = recoveryCallback;
}
/**
* Specify a {@link BiConsumer} for seeks management during
* {@link ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback)}
* call from the {@link org.springframework.kafka.listener.KafkaMessageListenerContainer}.
* This is called from the internal
* {@link org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter} implementation.
* @param onPartitionsAssignedCallback the {@link BiConsumer} to use
* @since 3.0.4
* @see ConsumerSeekAware#onPartitionsAssigned
*/
public void setOnPartitionsAssignedSeekCallback(
BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) {
this.onPartitionsAssignedSeekCallback = onPartitionsAssignedCallback;
}
@Override
protected void onInit() throws Exception {
super.onInit();
@@ -212,6 +234,13 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
super(null, null);
}
@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
if (KafkaInboundGateway.this.onPartitionsAssignedSeekCallback != null) {
KafkaInboundGateway.this.onPartitionsAssignedSeekCallback.accept(assignments, callback);
}
}
@Override
public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
Message<?> message = null;
@@ -254,7 +283,7 @@ public class KafkaInboundGateway<K, V, R> extends MessagingGatewaySupport implem
new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1);
if (message.getHeaders() instanceof KafkaMessageHeaders) {
((KafkaMessageHeaders) message.getHeaders()).getRawHeaders()
.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt);
.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt);
}
else {
messageToReturn = MessageBuilder.fromMessage(message)

90
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/inbound/KafkaMessageDrivenChannelAdapter.java
View File

@@ -17,10 +17,13 @@
package org.springframework.integration.kafka.inbound;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.BiConsumer;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.springframework.core.AttributeAccessor;
import org.springframework.integration.IntegrationMessageHeaderAccessor;
@@ -33,6 +36,7 @@ import org.springframework.integration.support.ErrorMessageUtils;
import org.springframework.integration.support.MessageBuilder;
import org.springframework.kafka.listener.AbstractMessageListenerContainer;
import org.springframework.kafka.listener.BatchMessageListener;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.listener.adapter.BatchMessagingMessageListenerAdapter;
import org.springframework.kafka.listener.adapter.FilteringBatchMessageListenerAdapter;
@@ -90,6 +94,8 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
private boolean filterInRetry;
private BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedSeekCallback;
/**
* Construct an instance with mode {@link ListenerMode#record}.
* @param messageListenerContainer the container.
@@ -225,6 +231,21 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
this.batchListener.setFallbackType(payloadType);
}
/**
* Specify a {@link BiConsumer} for seeks management during
* {@link ConsumerSeekAware.ConsumerSeekCallback#onPartitionsAssigned(Map, ConsumerSeekAware.ConsumerSeekCallback)}
* call from the {@link org.springframework.kafka.listener.KafkaMessageListenerContainer}.
* This is called from the internal
* {@link org.springframework.kafka.listener.adapter.MessagingMessageListenerAdapter} implementation.
* @param onPartitionsAssignedCallback the {@link BiConsumer} to use
* @since 3.0.4
* @see ConsumerSeekAware#onPartitionsAssigned
*/
public void setOnPartitionsAssignedSeekCallback(
BiConsumer<Map<TopicPartition, Long>, ConsumerSeekAware.ConsumerSeekCallback> onPartitionsAssignedCallback) {
this.onPartitionsAssignedSeekCallback = onPartitionsAssignedCallback;
}
@Override
public String getComponentType() {
return "kafka:message-driven-channel-adapter";
@@ -342,6 +363,23 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
}
}
private void sendMessageIfAny(Message<?> message, Object kafkaConsumedObject) {
if (message != null) {
try {
sendMessage(message);
}
finally {
if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) {
attributesHolder.remove();
}
}
}
else {
KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: "
+ kafkaConsumedObject);
}
}
/**
* The listener mode for the container, record or batch.
* @since 1.2
@@ -368,6 +406,13 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
super(null, null);
}
@Override
public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
if (KafkaMessageDrivenChannelAdapter.this.onPartitionsAssignedSeekCallback != null) {
KafkaMessageDrivenChannelAdapter.this.onPartitionsAssignedSeekCallback.accept(assignments, callback);
}
}
@Override
public void onMessage(ConsumerRecord<K, V> record, Acknowledgment acknowledgment, Consumer<?, ?> consumer) {
Message<?> message = null;
@@ -382,20 +427,8 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
RuntimeException exception = new ConversionException("Failed to convert to message for: " + record, e);
sendErrorMessageIfNecessary(null, exception);
}
if (message != null) {
try {
sendMessage(message);
}
finally {
if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) {
attributesHolder.remove();
}
}
}
else {
KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: "
+ record);
}
sendMessageIfAny(message, record);
}
private Message<?> addDeliveryAttemptHeader(Message<?> message) {
@@ -404,7 +437,7 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
new AtomicInteger(((RetryContext) attributesHolder.get()).getRetryCount() + 1);
if (message.getHeaders() instanceof KafkaMessageHeaders) {
((KafkaMessageHeaders) message.getHeaders()).getRawHeaders()
.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt);
.put(IntegrationMessageHeaderAccessor.DELIVERY_ATTEMPT, deliveryAttempt);
}
else {
messageToReturn = MessageBuilder.fromMessage(message)
@@ -444,8 +477,15 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
}
@Override
public void onMessage(List<ConsumerRecord<K, V>> records, Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
public void onPartitionsAssigned(Map<TopicPartition, Long> assignments, ConsumerSeekCallback callback) {
if (KafkaMessageDrivenChannelAdapter.this.onPartitionsAssignedSeekCallback != null) {
KafkaMessageDrivenChannelAdapter.this.onPartitionsAssignedSeekCallback.accept(assignments, callback);
}
}
@Override
public void onMessage(List<ConsumerRecord<K, V>> records, Acknowledgment acknowledgment,
Consumer<?, ?> consumer) {
Message<?> message = null;
try {
@@ -458,20 +498,8 @@ public class KafkaMessageDrivenChannelAdapter<K, V> extends MessageProducerSuppo
getMessagingTemplate().send(getErrorChannel(), new ErrorMessage(exception));
}
}
if (message != null) {
try {
sendMessage(message);
}
finally {
if (KafkaMessageDrivenChannelAdapter.this.retryTemplate == null) {
attributesHolder.remove();
}
}
}
else {
KafkaMessageDrivenChannelAdapter.this.logger.debug("Converter returned a null message for: "
+ records);
}
sendMessageIfAny(message, records);
}
@Override

9
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/dsl/KafkaDslTests.java
View File

@@ -210,6 +210,8 @@ public class KafkaDslTests {
this.kafkaTemplateTopic1.send(TEST_TOPIC3, "foo");
assertThat(this.config.sourceFlowLatch.await(10, TimeUnit.SECONDS)).isTrue();
assertThat(this.config.fromSource).isEqualTo("foo");
assertThat(this.config.onPartitionsAssignedCalledLatch.await(10, TimeUnit.SECONDS)).isTrue();
}
@Test
@@ -227,6 +229,8 @@ public class KafkaDslTests {
private final CountDownLatch replyContainerLatch = new CountDownLatch(1);
private final CountDownLatch onPartitionsAssignedCalledLatch = new CountDownLatch(1);
private Object fromSource;
@Bean
@@ -249,11 +253,14 @@ public class KafkaDslTests {
KafkaMessageDrivenChannelAdapter.ListenerMode.record, TEST_TOPIC1)
.configureListenerContainer(c ->
c.ackMode(ContainerProperties.AckMode.MANUAL)
.idleEventInterval(100L)
.id("topic1ListenerContainer"))
.recoveryCallback(new ErrorMessageSendingRecoverer(errorChannel(),
new RawRecordHeaderErrorMessageStrategy()))
.retryTemplate(new RetryTemplate())
.filterInRetry(true))
.filterInRetry(true)
.onPartitionsAssignedSeekCallback((map, callback) ->
ContextConfiguration.this.onPartitionsAssignedCalledLatch.countDown()))
.filter(Message.class, m ->
m.getHeaders().get(KafkaHeaders.RECEIVED_MESSAGE_KEY, Integer.class) < 101,
f -> f.throwExceptionOnRejection(true))

10
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/InboundGatewayTests.java
View File

@@ -23,6 +23,8 @@ import static org.springframework.kafka.test.assertj.KafkaConditions.value;
import java.lang.reflect.Type;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
@@ -105,6 +107,7 @@ public class InboundGatewayTests {
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic1);
containerProps.setIdleEventInterval(100L);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
@@ -133,6 +136,11 @@ public class InboundGatewayTests {
}
});
CountDownLatch onPartitionsAssignedCalledLatch = new CountDownLatch(1);
gateway.setOnPartitionsAssignedSeekCallback((map, seekConsumer) -> onPartitionsAssignedCalledLatch.countDown());
gateway.start();
ContainerTestUtils.waitForAssignment(container, 2);
@@ -149,11 +157,13 @@ public class InboundGatewayTests {
assertThat(headers.get(KafkaHeaders.TIMESTAMP_TYPE)).isEqualTo("CREATE_TIME");
assertThat(headers.get(KafkaHeaders.REPLY_TOPIC)).isEqualTo(topic2);
assertThat(headers.get("testHeader")).isEqualTo("testValue");
reply.send(MessageBuilder.withPayload("FOO").copyHeaders(headers).build());
ConsumerRecord<Integer, String> record = KafkaTestUtils.getSingleRecord(consumer, topic2);
assertThat(record).has(partition(1));
assertThat(record).has(value("FOO"));
assertThat(onPartitionsAssignedCalledLatch.await(10, TimeUnit.SECONDS)).isTrue();
gateway.stop();
}

8
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
View File

@@ -302,12 +302,18 @@ public class MessageDrivenAdapterTests {
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic2);
containerProps.setIdleEventInterval(100L);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
KafkaMessageDrivenChannelAdapter<Integer, String> adapter = new KafkaMessageDrivenChannelAdapter<>(container,
ListenerMode.batch);
QueueChannel out = new QueueChannel();
adapter.setOutputChannel(out);
final CountDownLatch onPartitionsAssignedCalledLatch = new CountDownLatch(1);
adapter.setOnPartitionsAssignedSeekCallback((map, consumer) -> onPartitionsAssignedCalledLatch.countDown());
adapter.afterPropertiesSet();
adapter.setBatchMessageConverter(new BatchMessagingMessageConverter() {
@@ -347,6 +353,8 @@ public class MessageDrivenAdapterTests {
.isEqualTo(Arrays.asList(1487694048607L, 1487694048608L));
assertThat(headers.get("testHeader")).isEqualTo("testValue");
assertThat(onPartitionsAssignedCalledLatch.await(10, TimeUnit.SECONDS)).isTrue();
adapter.setMessageConverter(new BatchMessageConverter() {
@Override

28
spring-integration-kafka/src/test/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandlerTests.java
View File

@@ -123,8 +123,8 @@ public class KafkaProducerMessageHandlerTests {
}
@Test
public void testOutbound() {
ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
public void testOutbound() throws Exception {
DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
@@ -174,11 +174,13 @@ public class KafkaProducerMessageHandlerTests {
assertThat(record).has(key(2));
assertThat(record).has(partition(1));
assertThat(record.value()).isNull();
producerFactory.destroy();
}
@Test
public void testOutboundWithTimestamp() {
ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
public void testOutboundWithTimestamp() throws Exception {
DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
@@ -203,11 +205,13 @@ public class KafkaProducerMessageHandlerTests {
new DefaultKafkaHeaderMapper().toHeaders(record.headers(), headers);
assertThat(headers.size()).isEqualTo(1);
assertThat(headers.get("baz")).isEqualTo("qux");
producerFactory.destroy();
}
@Test
public void testOutboundWithTimestampExpression() {
ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
public void testOutboundWithTimestampExpression() throws Exception {
DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
@@ -240,11 +244,13 @@ public class KafkaProducerMessageHandlerTests {
assertThat(record2).has(partition(1));
assertThat(record2).has(value("foo"));
assertThat(record2.timestamp()).isGreaterThanOrEqualTo(currentTimeMarker);
producerFactory.destroy();
}
@Test
public void testOutboundWithAsyncResults() {
ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
public void testOutboundWithAsyncResults() throws Exception {
DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(producerFactory);
KafkaProducerMessageHandler<Integer, String> handler = new KafkaProducerMessageHandler<>(template);
@@ -298,6 +304,8 @@ public class KafkaProducerMessageHandlerTests {
assertThat(((MessagingException) received.getPayload()).getFailedMessage()).isSameAs(message);
assertThat(((MessagingException) received.getPayload()).getCause()).isSameAs(fooException);
assertThat(((KafkaSendFailureException) received.getPayload()).getRecord()).isNotNull();
producerFactory.destroy();
}
@Test
@@ -322,7 +330,7 @@ public class KafkaProducerMessageHandlerTests {
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(consumerFactory, containerProperties);
ProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
DefaultKafkaProducerFactory<Integer, String> producerFactory = new DefaultKafkaProducerFactory<>(
KafkaTestUtils.producerProps(embeddedKafka));
ReplyingKafkaTemplate<Integer, String, String> template = new ReplyingKafkaTemplate<>(producerFactory, container);
template.start();
@@ -390,6 +398,8 @@ public class KafkaProducerMessageHandlerTests {
template.stop();
// discard from the test consumer
KafkaTestUtils.getSingleRecord(consumer, topic6);
producerFactory.destroy();
}
@SuppressWarnings({ "rawtypes", "unchecked" })