GH-3238: EmbeddedKafaka#kraft default to false

Fixes: #3238

Switch `EmbeddedKafaka#kraft` default to `false`.

spring-kafka-docs/src/main/antora/modules/ROOT/pages/testing.adoc

@@ -8,8 +8,8 @@ The `spring-kafka-test` jar contains some useful utilities to assist with testin
 
 Two implementations are provided:
 
-* `EmbeddedKafkaZKBroker` - legacy implementation which starts an embedded `Zookeeper` instance.
-* `EmbeddedKafkaKraftBroker` - (default) uses `Kraft` instead of `Zookeeper` in combined controller and broker modes (since 3.1).
+* `EmbeddedKafkaZKBroker` - legacy implementation which starts an embedded `Zookeeper` instance (which is still the default when using `EmbeddedKafka`).
+* `EmbeddedKafkaKraftBroker` - uses `Kraft` instead of `Zookeeper` in combined controller and broker modes (since 3.1).
 
 There are several techniques to configure the broker as discussed in the following sections.
 
@@ -228,7 +228,7 @@ In addition, these properties can be provided:
 - `spring.kafka.embedded.topics` - topics (comma-separated value) to create in the started Kafka cluster;
 - `spring.kafka.embedded.partitions` - number of partitions to provision for the created topics;
 - `spring.kafka.embedded.broker.properties.location` - the location of the file for additional Kafka broker configuration properties; the value of this property must follow the Spring resource abstraction pattern;
-- `spring.kafka.embedded.kraft` - when false, use an `EmbeddedKafkaZKBroker` instead of an `EmbeddedKafkaKraftBroker`.
+- `spring.kafka.embedded.kraft` - default false, when true, use an `EmbeddedKafkaKraftBroker` instead of an `EmbeddedKafkaZKBroker`.
 
 Essentially these properties mimic some of the `@EmbeddedKafka` attributes.
 
@@ -295,7 +295,7 @@ public class KafkaStreamsTests {
 
 Starting with version 2.2.4, you can also use the `@EmbeddedKafka` annotation to specify the Kafka ports property.
 
-Starting with version 3.1, set the `kraft` property to `false` to use an `EmbeddedKafkaZKBroker` instead of an `EmbeddedKafkaKraftBroker`.
+Starting with version 3.2, set the `kraft` property to `true` to use an `EmbeddedKafkaKraftBroker` instead of an `EmbeddedKafkaZKBroker`.
 
 The following example sets the `topics`, `brokerProperties`, and `brokerPropertiesLocation` attributes of `@EmbeddedKafka` support property placeholder resolutions:
 

spring-kafka-test/src/main/java/org/springframework/kafka/test/context/EmbeddedKafka.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2023 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -67,6 +67,7 @@ import org.springframework.test.context.aot.DisabledInAotMode;
  * @author Sergio Lourenco
  * @author Pawel Lozinski
  * @author Adrian Chlebosz
+ * @author Soby Chacko
  *
  * @since 1.3
  *
@@ -193,11 +194,11 @@ public @interface EmbeddedKafka {
 	int adminTimeout() default EmbeddedKafkaBroker.DEFAULT_ADMIN_TIMEOUT;
 
 	/**
-	 * Use KRaft instead of Zookeeper; default true.
+	 * Use KRaft instead of Zookeeper; default false.
 	 * @return whether to use KRaft.
 	 * @since 3.6
 	 */
-	boolean kraft() default true;
+	boolean kraft() default false;
 
 }
 

spring-kafka/src/test/java/org/springframework/kafka/annotation/EnableKafkaIntegrationTests.java

@@ -195,7 +195,7 @@ import jakarta.validation.constraints.Max;
 		"annotated29", "annotated30", "annotated30reply", "annotated31", "annotated32", "annotated33",
 		"annotated34", "annotated35", "annotated36", "annotated37", "foo", "manualStart", "seekOnIdle",
 		"annotated38", "annotated38reply", "annotated39", "annotated40", "annotated41", "annotated42",
-		"annotated43", "annotated43reply", "seekToComputeFn"})
+		"annotated43", "annotated43reply", "seekToComputeFn"}, kraft = true)
 @TestPropertySource(properties = "spel.props=fetch.min.bytes=420000,max.poll.records=10")
 public class EnableKafkaIntegrationTests {
 

spring-kafka/src/test/java/org/springframework/kafka/streams/KafkaStreamsInteractiveQueryServiceTests.java

@@ -67,7 +67,7 @@ import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.core.ProducerFactory;
 import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
 import org.springframework.kafka.test.EmbeddedKafkaBroker;
-import org.springframework.kafka.test.EmbeddedKafkaKraftBroker;
+import org.springframework.kafka.test.EmbeddedKafkaZKBroker;
 import org.springframework.kafka.test.context.EmbeddedKafka;
 import org.springframework.kafka.test.utils.KafkaTestUtils;
 import org.springframework.retry.RetryPolicy;
@@ -96,7 +96,7 @@ class KafkaStreamsInteractiveQueryServiceTests {
 	public static final String NON_EXISTENT_STORE = "my-non-existent-store";
 
 	@Autowired
-	private EmbeddedKafkaKraftBroker embeddedKafka;
+	private EmbeddedKafkaZKBroker embeddedKafka;
 
 	@Autowired
 	private StreamsBuilderFactoryBean streamsBuilderFactoryBean;

spring-kafka/src/test/java/org/springframework/kafka/streams/KafkaStreamsTests.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2017-2023 the original author or authors.
+ * Copyright 2017-2024 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.
@@ -97,7 +97,7 @@ import kafka.server.BrokerServer;
 		brokerProperties = {
 				"auto.create.topics.enable=${topics.autoCreate:false}",
 				"delete.topic.enable=${topic.delete:true}" },
-		brokerPropertiesLocation = "classpath:/${broker.filename:broker}.properties")
+		brokerPropertiesLocation = "classpath:/${broker.filename:broker}.properties", kraft = true)
 public class KafkaStreamsTests {
 
 	static final String STREAMING_TOPIC1 = "streamingTopic1";