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

GH-235: KPMH: Flush the template in stop()

Browse Source 
Resolves https://github.com/spring-projects/spring-integration-kafka/issues/235

**cherry-pick to 3.1.x**
This commit is contained in:
Gary Russell
2019-03-25 10:54:20 -04:00
committed by Artem Bilan
parent 4f54c4cab3
commit 71cdeee0b3
2 changed files with 29 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

24
spring-integration-kafka/src/main/java/org/springframework/integration/kafka/outbound/KafkaProducerMessageHandler.java
View File

@@ -27,6 +27,7 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
@@ -35,6 +36,7 @@ import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.springframework.context.Lifecycle;
import org.springframework.expression.EvaluationContext;
import org.springframework.expression.Expression;
import org.springframework.integration.MessageTimeoutException;
@@ -85,7 +87,8 @@ import org.springframework.util.concurrent.SettableListenableFuture;
*
* @since 0.5
*/
public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMessageHandler {
public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMessageHandler
implements Lifecycle {
private static final long DEFAULT_SEND_TIMEOUT = 10000;
@@ -97,6 +100,8 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
private final boolean transactional;
private final AtomicBoolean running = new AtomicBoolean();
private EvaluationContext evaluationContext;
private Expression topicExpression;
@@ -331,6 +336,23 @@ public class KafkaProducerMessageHandler<K, V> extends AbstractReplyProducingMes
this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory());
}
@Override
public void start() {
this.running.set(true);
}
@Override
public void stop() {
if (this.running.compareAndSet(true, false)) {
this.kafkaTemplate.flush();
}
}
@Override
public boolean isRunning() {
return this.running.get();
}
@SuppressWarnings("unchecked")
@Override
protected Object handleRequestMessage(final Message<?> message) {

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

@@ -1,5 +1,5 @@
/*
* Copyright 2016-2018 the original author or authors.
* Copyright 2016-2019 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.
@@ -414,11 +414,16 @@ public class KafkaProducerMessageHandlerTests {
KafkaTemplate template = new KafkaTemplate(pf);
KafkaProducerMessageHandler handler = new KafkaProducerMessageHandler(template);
handler.setTopicExpression(new LiteralExpression("bar"));
handler.setBeanFactory(mock(BeanFactory.class));
handler.afterPropertiesSet();
handler.start();
handler.handleMessage(new GenericMessage<>("foo"));
handler.stop();
InOrder inOrder = inOrder(producer);
inOrder.verify(producer).beginTransaction();
inOrder.verify(producer).send(any(ProducerRecord.class), any(Callback.class));
inOrder.verify(producer).commitTransaction();
inOrder.verify(producer).flush();
}
}