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

Fix tests for latest SK snapshot

Browse Source
This commit is contained in:
Gary Russell
2020-04-20 16:51:31 -04:00
committed by Artem Bilan
parent 04019dbb96
commit 8771f8e18c
1 changed files with 6 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -80,6 +80,7 @@ import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.listener.ContainerProperties.AssignmentCommitOption;
import org.springframework.kafka.listener.KafkaMessageListenerContainer;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.support.DefaultKafkaHeaderMapper;
@@ -503,12 +504,14 @@ class KafkaProducerMessageHandlerTests {
ConsumerFactory cf = mock(ConsumerFactory.class);
willReturn(mockConsumer).given(cf).createConsumer("group", "", null, KafkaTestUtils.defaultPropertyOverrides());
Producer producer = mock(Producer.class);
given(producer.send(any(), any())).willReturn(new SettableListenableFuture<>());
final CountDownLatch closeLatch = new CountDownLatch(2);
willAnswer(i -> {
closeLatch.countDown();
return null;
}).given(producer).close(any());
ProducerFactory pf = mock(ProducerFactory.class);
given(pf.isProducerPerConsumerPartition()).willReturn(true);
given(pf.transactionCapable()).willReturn(true);
final List<String> transactionalIds = new ArrayList<>();
willAnswer(i -> {
@@ -519,7 +522,7 @@ class KafkaProducerMessageHandlerTests {
ContainerProperties props = new ContainerProperties("foo");
props.setGroupId("group");
props.setTransactionManager(ptm);
props.setMissingTopicsFatal(false);
props.setAssignmentCommitOption(AssignmentCommitOption.ALWAYS);
final KafkaTemplate template = new KafkaTemplate(pf);
KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, props);
container.setBeanName("commit");
@@ -613,6 +616,7 @@ class KafkaProducerMessageHandlerTests {
ConsumerFactory cf = mock(ConsumerFactory.class);
willReturn(mockConsumer).given(cf).createConsumer("group", "", null, KafkaTestUtils.defaultPropertyOverrides());
Producer producer = mock(Producer.class);
given(producer.send(any(), any())).willReturn(new SettableListenableFuture<>());
final CountDownLatch closeLatch = new CountDownLatch(2);
willAnswer(i -> {
closeLatch.countDown();
@@ -637,7 +641,7 @@ class KafkaProducerMessageHandlerTests {
ContainerProperties props = new ContainerProperties("foo");
props.setGroupId("group");
props.setTransactionManager(tm);
props.setMissingTopicsFatal(false);
props.setAssignmentCommitOption(AssignmentCommitOption.ALWAYS);
final KafkaTemplate template = new KafkaTemplate(pf);
template.setTransactionIdPrefix("template.tx.id.");
KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, props);