From 10a764729c843db53e4e635b17c3b592a228f43c Mon Sep 17 00:00:00 2001
From: Artem Bilan <abilan@vmware.com>
Date: Sat, 18 Nov 2023 10:19:41 -0500
Subject: [PATCH] Fix `MessageDrivenAdapterTests` for partitions

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
---
 .../integration/kafka/inbound/MessageDrivenAdapterTests.java   | 3 ++-
 1 file changed, 2 insertions(+), 1 deletion(-)

diff --git a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
index 8bf0b64966..480d6c3dfc 100644
--- a/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
+++ b/spring-integration-kafka/src/test/java/org/springframework/integration/kafka/inbound/MessageDrivenAdapterTests.java
@@ -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 =