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

Fix MessageDrivenAdapterTests for partitions

Browse Source 
Related to https://github.com/spring-projects/spring-kafka/issues/2891

Starting with Spring for Apache Kafka `3.1`, the `ackMode` for
listener container is coerced to `MANUAL` if no `groupId` assigned
This commit is contained in:
Artem Bilan
2023-11-18 10:19:41 -05:00
parent 22c4db291b
commit 10a764729c
1 changed files with 2 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -579,7 +579,7 @@ class MessageDrivenAdapterTests {
void testPauseResume() throws Exception {
ConsumerFactory<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(isNull(), eq("clientId"), isNull(), any())).willReturn(consumer);
given(cf.createConsumer(eq("testPauseResumeGroup"), eq("clientId"), isNull(), any())).willReturn(consumer);
final Map<TopicPartition, List<ConsumerRecord<Integer, String>>> records = new HashMap<>();
records.put(new TopicPartition("foo", 0), Arrays.asList(
new ConsumerRecord<>("foo", 0, 0L, 1, "foo"),
@@ -602,6 +602,7 @@ class MessageDrivenAdapterTests {
ContainerProperties containerProps = new ContainerProperties(topicPartition);
containerProps.setAckMode(ContainerProperties.AckMode.RECORD);
containerProps.setClientId("clientId");
containerProps.setGroupId("testPauseResumeGroup");
containerProps.setIdleEventInterval(100L);
BlockingQueue<KafkaEvent> containerEvents = new LinkedBlockingQueue<>();
KafkaMessageListenerContainer<Integer, String> container =