Upgrade to Kafka 4.1.0 and update API usage

- Bump kafkaVersion from 4.0.0 to 4.1.0 in build.gradle
- Remove broker properties no longer needed in embedded Kafka tests:

  * unstable.api.versions.enable=true
  * group.coordinator.rebalance.protocols=classic,share

- Replace AcknowledgeType.ACCEPT with commitSync() in share consumer tests
    - Remove explicit acknowledge() calls in implicit mode tests since 4.1.0 client disallowed that
    - Leverage implicit acknowledgment where records are automatically treated as ACCEPT
    - Use commitSync() for explicit commit timing instead of relying on next poll() auto-commit
    - Maintains same test semantics while following Kafka 4.1 implicit mode best practices
- Update SerializationIntegrationTests to use Plugin-based deserializer access

- Update SerializationIntegrationTests to use Plugin-based deserializer access in 4.1.0 client
    - Replace direct deserializer access with plugin mechanism retrieval
    - Use KafkaTestUtils.getPropertyValue() to access valueDeserializerPlugin from consumer internals
    - Verify plugin returns expected delegating deserializer instance

Signed-off-by: Soby Chacko <[email protected]>

Author: sobychacko

build.gradle

@@ -60,7 +60,7 @@ ext {
 	jaywayJsonPathVersion = '2.9.0'
 	junit4Version = '4.13.2'
 	junitJupiterVersion = '5.13.4'
-	kafkaVersion = '4.0.0'
+	kafkaVersion = '4.1.0'
 	kotlinCoroutinesVersion = '1.10.2'
 	log4jVersion = '2.25.1'
 	micrometerDocsVersion = '1.0.4'

spring-kafka/src/test/java/org/springframework/kafka/core/DefaultShareConsumerFactoryTests.java

@@ -31,7 +31,6 @@
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.AlterConfigOp;
 import org.apache.kafka.clients.admin.ConfigEntry;
-import org.apache.kafka.clients.consumer.AcknowledgeType;
 import org.apache.kafka.clients.consumer.ShareConsumer;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerConfig;
@@ -56,8 +55,6 @@
 @EmbeddedKafka(
 		topics = {"embedded-share-test", "embedded-share-distribution-test"}, partitions = 1,
 		brokerProperties = {
-				"unstable.api.versions.enable=true",
-				"group.coordinator.rebalance.protocols=classic,share",
 				"share.coordinator.state.topic.replication.factor=1",
 				"share.coordinator.state.topic.min.isr=1"
 		})
@@ -248,7 +245,9 @@ private static List<String> runSharedConsumerTest(String topic, String groupId,
 						var records = consumer.poll(Duration.ofMillis(200));
 						for (var r : records) {
 							allReceived.add(r.value());
-							consumer.acknowledge(r, AcknowledgeType.ACCEPT);
+							// Leverage implicit acknowledgment where records are automatically treated as ACCEPT
+							// Use commitSync() for explicit commit timing instead of relying on next poll() auto-commit
+							consumer.commitSync(Duration.ofMillis(10000));
 							latch.countDown();
 						}
 					}

spring-kafka/src/test/java/org/springframework/kafka/listener/ShareKafkaListenerIntegrationTests.java

@@ -63,8 +63,6 @@
 @DirtiesContext
 @EmbeddedKafka(topics = "share-listener-integration-test",
 		brokerProperties = {
-				"unstable.api.versions.enable=true",
-				"group.coordinator.rebalance.protocols=classic,share",
 				"share.coordinator.state.topic.replication.factor=1",
 				"share.coordinator.state.topic.min.isr=1"
 		})

spring-kafka/src/test/java/org/springframework/kafka/listener/ShareKafkaMessageListenerContainerIntegrationTests.java

@@ -50,8 +50,6 @@
 @EmbeddedKafka(
 	topics = {"share-listener-integration-test"}, partitions = 1,
 	brokerProperties = {
-		"unstable.api.versions.enable=true",
-		"group.coordinator.rebalance.protocols=classic,share",
 		"share.coordinator.state.topic.replication.factor=1",
 		"share.coordinator.state.topic.min.isr=1"
 	}

spring-kafka/src/test/java/org/springframework/kafka/support/serializer/SerializationIntegrationTests.java

@@ -20,6 +20,7 @@
 import java.util.regex.Pattern;
 
 import org.apache.kafka.common.header.Headers;
+import org.apache.kafka.common.internals.Plugin;
 import org.apache.kafka.common.serialization.Deserializer;
 import org.apache.kafka.common.serialization.StringDeserializer;
 import org.junit.jupiter.api.Test;
@@ -37,6 +38,7 @@
 
 /**
  * @author Gary Russell
+ * @author Soby Chacko
  * @since 2.8.1
  *
  */
@@ -46,6 +48,7 @@ public class SerializationIntegrationTests {
 	public static final String DBTD_TOPIC = "dbtd";
 
 	@Test
+	@SuppressWarnings("unchecked")
 	void configurePreLoadedDelegates() {
 		Map<String, Object> consumerProps =
 				KafkaTestUtils.consumerProps(EmbeddedKafkaCondition.getBroker(), DBTD_TOPIC, false);
@@ -63,8 +66,11 @@ void configurePreLoadedDelegates() {
 		props.setMessageListener(mock(MessageListener.class));
 		KafkaMessageListenerContainer<String, Object> container = new KafkaMessageListenerContainer<>(cFact, props);
 		container.start();
-		assertThat(KafkaTestUtils.getPropertyValue(container,
-				"listenerConsumer.consumer.delegate.deserializers.valueDeserializer"))
+		//The Deserializers class uses a plugin mechanism to retrieve the actual deserializer.
+		Plugin<Deserializer<?>> valueDeserializerPlugin = (Plugin<Deserializer<?>>) KafkaTestUtils.getPropertyValue(container,
+				"listenerConsumer.consumer.delegate.deserializers.valueDeserializerPlugin");
+		assertThat(valueDeserializerPlugin).isNotNull();
+		assertThat(valueDeserializerPlugin.get())
 				.isSameAs(delegating);
 		Map<?, ?> delegates = KafkaTestUtils.getPropertyValue(delegating, "delegates", Map.class);
 		assertThat(delegates).hasSize(1);