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

KafkaTestUtils consumer properties API unification (#3938)

Browse Source 
* Deprecate `KafkaTestUtils.consumerProps` methods based on the `String` for `autoCommit`
* Introduce `KafkaTestUtils.consumerProps` based on the `boolean`  for `autoCommit`
* Fix respective deprecation warnings

Signed-off-by: Mikhail Polivakha <mikhailpolivakha@gmail.com>
This commit is contained in:
Mikhail Polivakha
2025-06-04 18:53:09 +03:00
committed by GitHub
parent 9674334eda
commit 05d649473e
49 changed files with 144 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java
View File

@@ -59,6 +59,7 @@ import org.springframework.util.Assert;
* @author Hugo Wood
* @author Artem Bilan
* @author Sanghyeok An
* @author Mikhail Polivakha
*/
public final class KafkaTestUtils {
@@ -81,13 +82,30 @@ public final class KafkaTestUtils {
* @param autoCommit the auto commit.
* @param embeddedKafka a {@link EmbeddedKafkaBroker} instance.
* @return the properties.
* @deprecated please, use {@link #consumerProps(EmbeddedKafkaBroker, String, boolean)} instead
*/
@Deprecated(forRemoval = true, since = "4.0.0")
public static Map<String, Object> consumerProps(String group, String autoCommit,
EmbeddedKafkaBroker embeddedKafka) {
return consumerProps(embeddedKafka.getBrokersAsString(), group, autoCommit);
}
/**
* Set up test properties for an {@code <Integer, String>} consumer.
*
* @param group the group id.
* @param autoCommit the auto commit.
* @param embeddedKafka a {@link EmbeddedKafkaBroker} instance.
* @return the properties.
* @since 4.0
*/
public static Map<String, Object> consumerProps(EmbeddedKafkaBroker embeddedKafka, String group,
boolean autoCommit) {
return consumerProps(embeddedKafka.getBrokersAsString(), group, autoCommit);
}
/**
* Set up test properties for an {@code <Integer, String>} consumer.
* @param brokers the bootstrapServers property.
@@ -96,7 +114,7 @@ public final class KafkaTestUtils {
* @since 3.3
*/
public static Map<String, Object> consumerProps(String brokers, String group) {
return consumerProps(brokers, group, "false");
return consumerProps(brokers, group, false);
}
/**
@@ -114,8 +132,22 @@ public final class KafkaTestUtils {
* @param group the group id.
* @param autoCommit the auto commit.
* @return the properties.
*/
* @deprecated Please, use {@link #consumerProps(String, String, boolean)} instead.
*/
@Deprecated(forRemoval = true, since = "4.0.0")
public static Map<String, Object> consumerProps(String brokers, String group, String autoCommit) {
return consumerProps(brokers, group, Boolean.parseBoolean(autoCommit));
}
/**
* Set up test properties for an {@code <Integer, String>} consumer.
* @param brokers the bootstrapServers property.
* @param group the group id.
* @param autoCommit the auto commit.
* @return the properties.
* @since 4.0
*/
public static Map<String, Object> consumerProps(String brokers, String group, boolean autoCommit) {
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokers);
props.put(ConsumerConfig.GROUP_ID_CONFIG, group);
@@ -235,7 +267,7 @@ public final class KafkaTestUtils {
public static ConsumerRecord<?, ?> getOneRecord(String brokerAddresses, String group, String topic, int partition,
boolean seekToLast, boolean commit, Duration timeout) {
Map<String, Object> consumerConfig = consumerProps(brokerAddresses, group, "false");
Map<String, Object> consumerConfig = consumerProps(brokerAddresses, group, false);
consumerConfig.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
try (KafkaConsumer consumer = new KafkaConsumer(consumerConfig)) {
TopicPartition topicPart = new TopicPartition(topic, partition);

4
spring-kafka-test/src/test/java/org/springframework/kafka/test/AddressableEmbeddedBrokerTests.java
View File

@@ -69,7 +69,7 @@ public class AddressableEmbeddedBrokerTests {
@Test
public void testLateStartedConsumer() {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(TEST_EMBEDDED, "false", this.broker);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.broker, TEST_EMBEDDED, false);
Consumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
this.broker.consumeFromAnEmbeddedTopic(consumer, TEST_EMBEDDED);
@@ -78,7 +78,7 @@ public class AddressableEmbeddedBrokerTests {
producer.close();
KafkaTestUtils.getSingleRecord(consumer, TEST_EMBEDDED);
consumerProps = KafkaTestUtils.consumerProps("another" + TEST_EMBEDDED, "false", this.broker);
consumerProps = KafkaTestUtils.consumerProps(this.broker, "another" + TEST_EMBEDDED, false);
Consumer<Integer, String> consumer2 = new KafkaConsumer<>(consumerProps);
this.broker.consumeFromAnEmbeddedTopic(consumer2, TEST_EMBEDDED);
KafkaTestUtils.getSingleRecord(consumer2, TEST_EMBEDDED);

2
spring-kafka-test/src/test/java/org/springframework/kafka/test/EmbeddedKafkaKraftBrokerTests.java
View File

@@ -51,7 +51,7 @@ public class EmbeddedKafkaKraftBrokerTests {
Map<String, Object> producerProps = KafkaTestUtils.producerProps(kafka);
KafkaProducer<Integer, String> producer = new KafkaProducer<>(producerProps);
producer.send(new ProducerRecord<>("seekTestTopic", 0, 1, "beforeSeekToEnd"));
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("seekTest", "false", kafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(kafka, "seekTest", false);
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
kafka.consumeFromAnEmbeddedTopic(consumer, true /* seekToEnd */, "seekTestTopic");
producer.send(new ProducerRecord<>("seekTestTopic", 0, 1, "afterSeekToEnd"));

6
spring-kafka-test/src/test/java/org/springframework/kafka/test/utils/KafkaTestUtilsTests.java
View File

@@ -55,7 +55,7 @@ public class KafkaTestUtilsTests {
producer.send(new ProducerRecord<>("singleTopic1", 0, 1, "foo"));
producer.send(new ProducerRecord<>("singleTopic2", 0, 1, "foo"));
producer.close();
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("ktuTests1", "false", broker);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(broker, "ktuTests1", false);
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
broker.consumeFromAllEmbeddedTopics(consumer);
KafkaTestUtils.getSingleRecord(consumer, "singleTopic1");
@@ -72,7 +72,7 @@ public class KafkaTestUtilsTests {
Map<String, Object> producerProps = KafkaTestUtils.producerProps(broker);
KafkaProducer<Integer, String> producer = new KafkaProducer<>(producerProps);
producer.send(new ProducerRecord<>("singleTopic4", 0, 1, "foo"));
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("ktuTests2", "false", broker);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(broker, "ktuTests2", false);
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
broker.consumeFromEmbeddedTopics(consumer, "singleTopic4", "singleTopic5");
long t1 = System.currentTimeMillis();
@@ -117,7 +117,7 @@ public class KafkaTestUtilsTests {
Map<String, Object> producerProps = KafkaTestUtils.producerProps(broker);
KafkaProducer<Integer, String> producer = new KafkaProducer<>(producerProps);
producer.send(new ProducerRecord<>("multiTopic1", 0, 1, "foo"));
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("ktuTests3", "false", broker);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(broker, "ktuTests3", false);
KafkaConsumer<Integer, String> consumer = new KafkaConsumer<>(consumerProps);
broker.consumeFromAnEmbeddedTopic(consumer, "multiTopic1");
new Thread(() -> {

4
spring-kafka/src/test/java/org/springframework/kafka/annotation/AliasPropertiesTests.java
View File

@@ -157,9 +157,7 @@ public class AliasPropertiesTests {
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> consumerProps =
KafkaTestUtils.consumerProps("myAliasGroup", "false", embeddedKafka());
return consumerProps;
return KafkaTestUtils.consumerProps(embeddedKafka(), "myAliasGroup", false);
}
@Bean

2
spring-kafka/src/test/java/org/springframework/kafka/annotation/AsyncListenerTests.java
View File

@@ -343,7 +343,7 @@ public class AsyncListenerTests {
@Bean
public Map<String, Object> consumerConfigs(EmbeddedKafkaBroker embeddedKafka) {
return KafkaTestUtils.consumerProps("test", "false", embeddedKafka);
return KafkaTestUtils.consumerProps(embeddedKafka, "test", false);
}
@Bean

2
spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversion2Tests.java
View File

@@ -105,7 +105,7 @@ public class BatchListenerConversion2Tests {
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> consumerProps =
KafkaTestUtils.consumerProps(DEFAULT_TEST_GROUP_ID, "false", this.embeddedKafka);
KafkaTestUtils.consumerProps(this.embeddedKafka, DEFAULT_TEST_GROUP_ID, false);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
consumerProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, JsonDeserializer.class);
consumerProps.put(ErrorHandlingDeserializer.VALUE_FUNCTION, FailedFooProvider.class);

2
spring-kafka/src/test/java/org/springframework/kafka/annotation/BatchListenerConversionTests.java
View File

@@ -199,7 +199,7 @@ public class BatchListenerConversionTests {
@Bean
public Map<String, Object> consumerConfigs(EmbeddedKafkaBroker embeddedKafka) {
Map<String, Object> consumerProps =
KafkaTestUtils.consumerProps(DEFAULT_TEST_GROUP_ID, "false", embeddedKafka);
KafkaTestUtils.consumerProps(embeddedKafka, DEFAULT_TEST_GROUP_ID, false);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, BytesDeserializer.class);
consumerProps.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1000);
consumerProps.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);

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

@@ -1555,7 +1555,7 @@ public class EnableKafkaIntegrationTests {
@Bean
public Map<String, Object> consumerConfigs() {
return KafkaTestUtils.consumerProps(DEFAULT_TEST_GROUP_ID, "false", this.embeddedKafka);
return KafkaTestUtils.consumerProps(this.embeddedKafka, DEFAULT_TEST_GROUP_ID, false);
}
@Bean

8
spring-kafka/src/test/java/org/springframework/kafka/core/DefaultKafkaConsumerFactoryTests.java
View File

@@ -367,7 +367,7 @@ public class DefaultKafkaConsumerFactoryTests {
DefaultKafkaProducerFactory<Integer, String> pfTx = new DefaultKafkaProducerFactory<>(producerProps);
pfTx.setTransactionIdPrefix("fooTx.");
KafkaOperations<Integer, String> templateTx = new KafkaTemplate<>(pfTx);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("txCache1Group", "false", this.embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.embeddedKafka, "txCache1Group", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
AtomicReference<Consumer<Integer, String>> wrapped = new AtomicReference<>();
cf.addPostProcessor(consumer -> {
@@ -418,7 +418,7 @@ public class DefaultKafkaConsumerFactoryTests {
TransactionIdSuffixStrategy suffixStrategy = new DefaultTransactionIdSuffixStrategy(3);
pfTx.setTransactionIdSuffixStrategy(suffixStrategy);
KafkaOperations<Integer, String> templateTx = new KafkaTemplate<>(pfTx);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("txCache1FixedGroup", "false", this.embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.embeddedKafka, "txCache1FixedGroup", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
AtomicReference<Consumer<Integer, String>> wrapped = new AtomicReference<>();
cf.addPostProcessor(consumer -> {
@@ -464,7 +464,7 @@ public class DefaultKafkaConsumerFactoryTests {
@ParameterizedTest
@ValueSource(booleans = { true, false })
void listener(boolean closeWithTimeout) {
Map<String, Object> consumerConfig = KafkaTestUtils.consumerProps("txCache1Group", "false", this.embeddedKafka);
Map<String, Object> consumerConfig = KafkaTestUtils.consumerProps(this.embeddedKafka, "txCache1Group", false);
consumerConfig.put(ConsumerConfig.CLIENT_ID_CONFIG, "foo-0");
DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(consumerConfig);
List<String> adds = new ArrayList<>();
@@ -503,7 +503,7 @@ public class DefaultKafkaConsumerFactoryTests {
void configDeserializer() {
Deserializer key = mock(Deserializer.class);
Deserializer value = mock(Deserializer.class);
Map<String, Object> config = KafkaTestUtils.consumerProps("mockGroup", "false", this.embeddedKafka);
Map<String, Object> config = KafkaTestUtils.consumerProps(this.embeddedKafka, "mockGroup", false);
DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory(config, key, value);
Deserializer keyDeserializer = cf.getKeyDeserializer();
assertThat(keyDeserializer).isSameAs(key);

8
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTests.java
View File

@@ -136,7 +136,7 @@ public class KafkaTemplateTests {
public static void setUp() {
embeddedKafka = EmbeddedKafkaCondition.getBroker();
Map<String, Object> consumerProps = KafkaTestUtils
.consumerProps("KafkaTemplatetests" + UUID.randomUUID(), "false", embeddedKafka);
.consumerProps(embeddedKafka, "KafkaTemplatetests" + UUID.randomUUID(), false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
consumer = cf.createConsumer();
embeddedKafka.consumeFromAnEmbeddedTopic(consumer, INT_KEY_TOPIC);
@@ -165,7 +165,7 @@ public class KafkaTemplateTests {
template.setDefaultTopic(INT_KEY_TOPIC);
template.setConsumerFactory(
new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps("xx", "false", embeddedKafka)));
new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps(embeddedKafka, "xx", false)));
ConsumerRecords<Integer, String> initialRecords =
template.receive(Collections.singleton(new TopicPartitionOffset(INT_KEY_TOPIC, 1, 1L)));
assertThat(initialRecords).isEmpty();
@@ -475,7 +475,7 @@ public class KafkaTemplateTests {
pf.setKeySerializer(new StringSerializer());
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf, true);
template.setDefaultTopic(STRING_KEY_TOPIC);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testTString", "false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "testTString", false);
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
cf.setKeyDeserializer(new StringDeserializer());
Consumer<String, String> localConsumer = cf.createConsumer();
@@ -630,7 +630,7 @@ public class KafkaTemplateTests {
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf, true);
template.setConsumerFactory(
new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps("xx", "false", embeddedKafka)));
new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps(embeddedKafka, "xx", false)));
TopicPartitionOffset tpoWithNullOffset = new TopicPartitionOffset(INT_KEY_TOPIC, 1, offset);
assertThatExceptionOfType(KafkaException.class)

6
spring-kafka/src/test/java/org/springframework/kafka/core/KafkaTemplateTransactionTests.java
View File

@@ -115,7 +115,7 @@ public class KafkaTemplateTransactionTests {
pf.setKeySerializer(new StringSerializer());
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(STRING_KEY_TOPIC);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testLocalTx", "false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "testLocalTx", false);
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
cf.setKeyDeserializer(new StringDeserializer());
@@ -173,7 +173,7 @@ public class KafkaTemplateTransactionTests {
pf.setTransactionIdSuffixStrategy(suffixStrategy);
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(STRING_KEY_TOPIC);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testLocalTxFixed", "false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "testLocalTxFixed", false);
consumerProps.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
cf.setKeyDeserializer(new StringDeserializer());
@@ -232,7 +232,7 @@ public class KafkaTemplateTransactionTests {
pf.setTransactionIdPrefix("my.transaction.");
KafkaTemplate<String, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(STRING_KEY_TOPIC);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("testGlobalTx", "false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, "testGlobalTx", false);
DefaultKafkaConsumerFactory<String, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
cf.setKeyDeserializer(new StringDeserializer());
Consumer<String, String> consumer = cf.createConsumer();

2
spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateIntegrationTests.java
View File

@@ -99,7 +99,7 @@ public class ReactiveKafkaProducerTemplateIntegrationTests {
@BeforeAll
public static void setUpBeforeClass() {
Map<String, Object> consumerProps = KafkaTestUtils
.consumerProps("reactive_consumer_group", "false", EmbeddedKafkaCondition.getBroker());
.consumerProps(EmbeddedKafkaCondition.getBroker(), "reactive_consumer_group", false);
reactiveKafkaConsumerTemplate =
new ReactiveKafkaConsumerTemplate<>(setupReceiverOptionsWithDefaultTopic(consumerProps));
}

2
spring-kafka/src/test/java/org/springframework/kafka/core/reactive/ReactiveKafkaProducerTemplateTransactionIntegrationTests.java
View File

@@ -100,7 +100,7 @@ public class ReactiveKafkaProducerTemplateTransactionIntegrationTests {
@BeforeAll
public static void setUpBeforeClass() {
Map<String, Object> consumerProps =
KafkaTestUtils.consumerProps(CONSUMER_GROUP_ID, "false", EmbeddedKafkaCondition.getBroker());
KafkaTestUtils.consumerProps(EmbeddedKafkaCondition.getBroker(), CONSUMER_GROUP_ID, false);
reactiveKafkaConsumerTemplate =
new ReactiveKafkaConsumerTemplate<>(setupReceiverOptionsWithDefaultTopic(consumerProps));
}

2
spring-kafka/src/test/java/org/springframework/kafka/listener/AbstractConsumerSeekAwareTests.java
View File

@@ -150,7 +150,7 @@ class AbstractConsumerSeekAwareTests {
@Bean
ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps("test-group", "false", this.broker));
return new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps(this.broker, "test-group", false));
}
@Bean

2
spring-kafka/src/test/java/org/springframework/kafka/listener/AsyncAckAfterHandleTests.java
View File

@@ -110,7 +110,7 @@ public class AsyncAckAfterHandleTests {
@Bean
ConsumerFactory<Integer, String> consumerFactory(EmbeddedKafkaBroker broker) {
Map<String, Object> props = KafkaTestUtils.consumerProps("asaac.grp", "false", broker);
Map<String, Object> props = KafkaTestUtils.consumerProps(broker, "asaac.grp", false);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 3);
return new DefaultKafkaConsumerFactory<>(
props);

26
spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java
View File

@@ -130,7 +130,7 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testAutoCommit() throws Exception {
this.logger.info("Start auto");
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test1", true);
AtomicReference<Properties> overrides = new AtomicReference<>();
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {
@@ -253,7 +253,7 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testAutoCommitWithRebalanceListener() throws Exception {
this.logger.info("Start auto");
Map<String, Object> props = KafkaTestUtils.consumerProps("test10", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test10", false);
AtomicReference<Properties> overrides = new AtomicReference<>();
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {
@@ -330,7 +330,7 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testAfterListenCommit() throws Exception {
this.logger.info("Start manual");
Map<String, Object> props = KafkaTestUtils.consumerProps("test2", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test2", false);
props.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
AtomicReference<Properties> overrides = new AtomicReference<>();
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {
@@ -406,7 +406,7 @@ public class ConcurrentMessageListenerContainerTests {
private void testManualCommitGuts(ContainerProperties.AckMode ackMode, String topic, int qual) throws Exception {
this.logger.info("Start " + ackMode);
Map<String, Object> props = KafkaTestUtils.consumerProps("test" + ackMode + qual, "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test" + ackMode + qual, false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic);
final CountDownLatch latch = new CountDownLatch(4);
@@ -451,7 +451,7 @@ public class ConcurrentMessageListenerContainerTests {
template.sendDefault(0, "baz");
template.sendDefault(2, "qux");
template.flush();
Map<String, Object> props = KafkaTestUtils.consumerProps("testManualExisting", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testManualExisting", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic7);
final CountDownLatch latch = new CountDownLatch(8);
@@ -502,7 +502,7 @@ public class ConcurrentMessageListenerContainerTests {
template.sendDefault(0, "baz");
template.sendDefault(2, "qux");
template.flush();
Map<String, Object> props = KafkaTestUtils.consumerProps("testManualExistingSync", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testManualExistingSync", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
ContainerProperties containerProps = new ContainerProperties(topic8);
containerProps.setSyncCommits(true);
@@ -540,7 +540,7 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testPausedStart() throws Exception {
this.logger.info("Start paused start");
Map<String, Object> props = KafkaTestUtils.consumerProps("test12", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test12", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic12);
@@ -630,7 +630,7 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testListenerException() throws Exception {
this.logger.info("Start exception");
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test1", true);
props.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic6);
@@ -682,7 +682,7 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testAckOnErrorRecord() throws Exception {
logger.info("Start ack on error");
Map<String, Object> props = KafkaTestUtils.consumerProps("test9", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test9", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
final CountDownLatch latch = new CountDownLatch(4);
ContainerProperties containerProps = new ContainerProperties(topic9);
@@ -766,7 +766,7 @@ public class ConcurrentMessageListenerContainerTests {
private void testAckOnErrorWithManualImmediateGuts(String topic, boolean ackOnError) throws Exception {
logger.info("Start ack on error with ManualImmediate ack mode");
Map<String, Object> props = KafkaTestUtils.consumerProps("testMan" + ackOnError, "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testMan" + ackOnError, false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
final CountDownLatch latch = new CountDownLatch(2);
ContainerProperties containerProps = new ContainerProperties(topic);
@@ -826,8 +826,7 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testIsChildRunning() throws Exception {
this.logger.info("Start isChildRunning");
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "true",
embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test1", true);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props) {
@Override
@@ -974,8 +973,7 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testContainerStartStop() throws Exception {
this.logger.info("Start containerStartStop");
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "true",
embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test1", true);
AtomicReference<Properties> overrides = new AtomicReference<>();
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {

2
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerEnforceRebalanceTests.java
View File

@@ -135,7 +135,7 @@ public class ContainerEnforceRebalanceTests {
@Bean
ConsumerFactory<Integer, String> cf() {
return new DefaultKafkaConsumerFactory<>(
KafkaTestUtils.consumerProps("enforce-rebalance-topic", "false", this.broker));
KafkaTestUtils.consumerProps(this.broker, "enforce-rebalance-topic", false));
}
}

2
spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerGroupSequencerTests.java
View File

@@ -154,7 +154,7 @@ public class ContainerGroupSequencerTests {
@Bean
ConsumerFactory<Integer, String> cf(EmbeddedKafkaBroker broker) {
return new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps("", "false", broker));
return new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps(broker, "", false));
}
@Bean

4
spring-kafka/src/test/java/org/springframework/kafka/listener/DefaultErrorHandlerBatchIntegrationTests.java
View File

@@ -76,7 +76,7 @@ public class DefaultErrorHandlerBatchIntegrationTests {
@Test
public void recoveryAndDlt() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("recoverBatch", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "recoverBatch", false);
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1000);
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -147,7 +147,7 @@ public class DefaultErrorHandlerBatchIntegrationTests {
@Test
public void recoveryFails() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("recoverBatch2", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "recoverBatch2", false);
props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 1000);
props.put(ConsumerConfig.FETCH_MAX_WAIT_MS_CONFIG, 500);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);

2
spring-kafka/src/test/java/org/springframework/kafka/listener/DeliveryAttemptAwareRetryListenerIntegrationTests.java
View File

@@ -246,7 +246,7 @@ class DeliveryAttemptAwareRetryListenerIntegrationTests {
Map<String, Object> props = KafkaTestUtils.consumerProps(
this.broker.getBrokersAsString(),
"DeliveryAttemptAwareRetryListenerIntegrationTestsGroupId",
"true");
true);
return new DefaultKafkaConsumerFactory<>(props);
}

4
spring-kafka/src/test/java/org/springframework/kafka/listener/ErrorHandlingDeserializerTests.java
View File

@@ -245,7 +245,7 @@ public class ErrorHandlingDeserializerTests {
@Bean
public ConsumerFactory<String, String> cf() {
Map<String, Object> props = KafkaTestUtils.consumerProps(TOPIC + ".g1", "false", embeddedKafka());
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka(), TOPIC + ".g1", false);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ExtendedEHD.class.getName());
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
props.put(ErrorHandlingDeserializer.KEY_DESERIALIZER_CLASS, FailSometimesDeserializer.class);
@@ -255,7 +255,7 @@ public class ErrorHandlingDeserializerTests {
@Bean
public ConsumerFactory<String, String> cfWithExplicitDeserializers() {
Map<String, Object> props = KafkaTestUtils.consumerProps(TOPIC + ".g2", "false", embeddedKafka());
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka(), TOPIC + ".g2", false);
return new DefaultKafkaConsumerFactory<>(props,
new ErrorHandlingDeserializer<String>(new FailSometimesDeserializer()).keyDeserializer(true),
new ErrorHandlingDeserializer<String>(new FailSometimesDeserializer()));

4
spring-kafka/src/test/java/org/springframework/kafka/listener/FallbackBatchErrorHandlerIntegrationTests.java
View File

@@ -78,7 +78,7 @@ public class FallbackBatchErrorHandlerIntegrationTests {
@Test
public void testRetriesAndDlt() throws InterruptedException {
Map<String, Object> props = KafkaTestUtils.consumerProps("retryBatch", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "retryBatch", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic1);
containerProps.setPollTimeout(10_000);
@@ -163,7 +163,7 @@ public class FallbackBatchErrorHandlerIntegrationTests {
@Test
public void testRetriesCantRecover() throws InterruptedException {
Map<String, Object> props = KafkaTestUtils.consumerProps("retryBatch2", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "retryBatch2", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic2);
containerProps.setPollTimeout(10_000);

52
spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
View File

@@ -222,7 +222,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testDelegateType() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("delegate", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "delegate", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic3);
containerProps.setShutdownTimeout(60_000L);
@@ -308,7 +308,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testNoResetPolicy() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("delegate", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "delegate", false);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "none");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic17);
@@ -334,7 +334,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testListenerTypes() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("lt1", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "lt1", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic4);
@@ -475,7 +475,7 @@ public class KafkaMessageListenerContainerTests {
@SuppressWarnings("unchecked")
@Test
public void testCommitsAreFlushedOnStop() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("flushedOnStop", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "flushedOnStop", false);
DefaultKafkaConsumerFactory<Integer, String> cf = spy(new DefaultKafkaConsumerFactory<>(props));
AtomicReference<Consumer<Integer, String>> consumer = new AtomicReference<>();
willAnswer(inv -> {
@@ -526,7 +526,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testRecordAck() throws Exception {
logger.info("Start record ack");
Map<String, Object> props = KafkaTestUtils.consumerProps("test6", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test6", false);
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 1);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic6);
@@ -1149,7 +1149,7 @@ public class KafkaMessageListenerContainerTests {
public void testBatchAck() throws Exception {
logger.info("Start batch ack");
Map<String, Object> props = KafkaTestUtils.consumerProps("test6", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test6", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic7);
containerProps.setMessageListener((MessageListener<Integer, String>) message -> {
@@ -1216,7 +1216,7 @@ public class KafkaMessageListenerContainerTests {
public void testBatchListener() throws Exception {
logger.info("Start batch listener");
Map<String, Object> props = KafkaTestUtils.consumerProps("test8", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test8", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic8);
containerProps.setMessageListener((BatchMessageListener<Integer, String>) messages -> {
@@ -1293,7 +1293,7 @@ public class KafkaMessageListenerContainerTests {
template.sendDefault(1, 0, "qux");
template.flush();
Map<String, Object> props = KafkaTestUtils.consumerProps("test9", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test9", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic9);
final CountDownLatch latch = new CountDownLatch(4);
@@ -1353,7 +1353,7 @@ public class KafkaMessageListenerContainerTests {
public void testBatchListenerErrors() throws Exception {
logger.info("Start batch listener errors");
Map<String, Object> props = KafkaTestUtils.consumerProps("test9", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test9", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic10);
containerProps.setMessageListener((BatchMessageListener<Integer, String>) messages -> {
@@ -1484,7 +1484,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testSeekBatch() throws Exception {
logger.info("Start seek batch seek");
Map<String, Object> props = KafkaTestUtils.consumerProps("test16", "true", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test16", true);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic16);
final CountDownLatch registerLatch = new CountDownLatch(1);
@@ -1538,11 +1538,11 @@ public class KafkaMessageListenerContainerTests {
}
private static Stream<Arguments> testSeekParameters() {
Map<String, Object> noAutoCommit = KafkaTestUtils.consumerProps("test15", "true", embeddedKafka);
Map<String, Object> noAutoCommit = KafkaTestUtils.consumerProps(embeddedKafka, "test15", true);
noAutoCommit.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); // test false by default
return Stream.of(
Arguments.of(KafkaTestUtils.consumerProps("test11", "false", embeddedKafka), topic11, false),
Arguments.of(KafkaTestUtils.consumerProps("test12", "true", embeddedKafka), topic12, true),
Arguments.of(KafkaTestUtils.consumerProps(embeddedKafka, "test11", false), topic11, false),
Arguments.of(KafkaTestUtils.consumerProps(embeddedKafka, "test12", true), topic12, true),
Arguments.of(noAutoCommit, topic15, false));
}
@@ -1675,7 +1675,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testDefinedPartitions() throws Exception {
this.logger.info("Start defined parts");
Map<String, Object> props = KafkaTestUtils.consumerProps("test13", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test13", false);
TopicPartitionOffset topic1Partition0 = new TopicPartitionOffset(topic13, 0, 0L);
CountDownLatch initialConsumersLatch = new CountDownLatch(2);
@@ -1986,7 +1986,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testManualAckRebalance() throws Exception {
logger.info("Start manual ack rebalance");
Map<String, Object> props = KafkaTestUtils.consumerProps("test14", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test14", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic14);
final List<AtomicInteger> counts = new ArrayList<>();
@@ -2078,7 +2078,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testJsonSerDeConfiguredType() throws Exception {
this.logger.info("Start JSON1");
Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testJson", false);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, Foo.class);
DefaultKafkaConsumerFactory<Integer, Foo> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -2118,7 +2118,7 @@ public class KafkaMessageListenerContainerTests {
public void testJsonSerDeWithInstanceDoesNotUseConfiguration() throws Exception {
this.logger.info("Start JSON1a");
Class<Foo1> consumerConfigValueDefaultType = Foo1.class;
Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testJson", false);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
props.put(JsonDeserializer.VALUE_DEFAULT_TYPE, consumerConfigValueDefaultType);
DefaultKafkaConsumerFactory<Integer, Foo> cf = new DefaultKafkaConsumerFactory<>(props, null, new JsonDeserializer<>(Foo.class));
@@ -2158,7 +2158,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testJsonSerDeHeaderSimpleType() throws Exception {
this.logger.info("Start JSON2");
Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testJson", false);
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
@@ -2200,7 +2200,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testJsonSerDeTypeMappings() throws Exception {
this.logger.info("Start JSON3");
Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testJson", false);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, JsonDeserializer.class);
props.put(JsonDeserializer.TRUSTED_PACKAGES, "*");
props.put(JsonDeserializer.TYPE_MAPPINGS, "foo:" + Foo1.class.getName() + " , bar:" + Bar1.class.getName());
@@ -2241,7 +2241,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testJsonSerDeIgnoreTypeHeadersInbound() throws Exception {
this.logger.info("Start JSON4");
Map<String, Object> props = KafkaTestUtils.consumerProps("testJson", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testJson", false);
props.put("spring.deserializer.value.delegate.class",
"org.apache.kafka.common.serialization.StringDeserializer");
ErrorHandlingDeserializer<Foo1> errorHandlingDeserializer =
@@ -2283,7 +2283,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testStaticAssign() throws Exception {
this.logger.info("Start static");
Map<String, Object> props = KafkaTestUtils.consumerProps("testStatic", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testStatic", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(new TopicPartitionOffset(topic22, 0),
@@ -2326,7 +2326,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testPatternAssign() throws Exception {
this.logger.info("Start pattern");
Map<String, Object> props = KafkaTestUtils.consumerProps("testpattern", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testpattern", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(Pattern.compile(topic23 + ".*"));
@@ -2359,7 +2359,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testBadListenerType() {
Map<String, Object> props = KafkaTestUtils.consumerProps("testStatic", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testStatic", false);
DefaultKafkaConsumerFactory<Integer, Foo1> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties("foo");
containerProps.setMissingTopicsFatal(false);
@@ -2381,7 +2381,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testBadAckMode() {
Map<String, Object> props = KafkaTestUtils.consumerProps("testStatic", "true", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testStatic", true);
DefaultKafkaConsumerFactory<Integer, Foo1> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties("foo");
containerProps.setMissingTopicsFatal(false);
@@ -2398,7 +2398,7 @@ public class KafkaMessageListenerContainerTests {
@Test
public void testRebalanceAfterFailedRecord() throws Exception {
logger.info("Start rebalance after failed record");
Map<String, Object> props = KafkaTestUtils.consumerProps("test18", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test18", false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic18);
final List<AtomicInteger> counts = new ArrayList<>();
@@ -3003,7 +3003,7 @@ public class KafkaMessageListenerContainerTests {
final CountDownLatch consumeFirstLatch = new CountDownLatch(1);
final CountDownLatch consumeLatch = new CountDownLatch(2);
Map<String, Object> props = KafkaTestUtils.consumerProps("test19", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "test19", false);
props.put(ConsumerConfig.MAX_POLL_INTERVAL_MS_CONFIG, 3_000);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic19);

4
spring-kafka/src/test/java/org/springframework/kafka/listener/MissingTopicCheckOverrideAdminConfigTests.java
View File

@@ -47,7 +47,7 @@ public class MissingTopicCheckOverrideAdminConfigTests {
@Test
void configOverride(EmbeddedKafkaBroker broker) {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("grp", "false", broker);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(broker, "grp", false);
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "junkjunk");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
ContainerProperties props = new ContainerProperties("mtccac");
@@ -70,7 +70,7 @@ public class MissingTopicCheckOverrideAdminConfigTests {
@Test
void configOverrideDefault(EmbeddedKafkaBroker broker) {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("grp", "false", broker);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(broker, "grp", false);
consumerProps.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "junkjunk");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
ContainerProperties props = new ContainerProperties("mtccac");

4
spring-kafka/src/test/java/org/springframework/kafka/listener/MissingTopicsTests.java
View File

@@ -47,7 +47,7 @@ public class MissingTopicsTests {
@Test
public void testMissingTopicCMLC() {
Map<String, Object> props = KafkaTestUtils.consumerProps("missing1", "true", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "missing1", true);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties("notexisting");
containerProps.setMessageListener((MessageListener<Integer, String>) message -> { });
@@ -67,7 +67,7 @@ public class MissingTopicsTests {
@Test
public void testMissingTopicKMLC() {
Map<String, Object> props = KafkaTestUtils.consumerProps("missing2", "true", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "missing2", true);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties("notexisting");
containerProps.setMessageListener((MessageListener<Integer, String>) message -> { });

2
spring-kafka/src/test/java/org/springframework/kafka/listener/PauseContainerWhileErrorHandlerIsRetryingTests.java
View File

@@ -213,7 +213,7 @@ public class PauseContainerWhileErrorHandlerIsRetryingTests {
@Bean
ConcurrentKafkaListenerContainerFactory kafkaListenerContainerFactory() {
DefaultKafkaConsumerFactory consumerFactory = new DefaultKafkaConsumerFactory(
KafkaTestUtils.consumerProps("grp", "false", embeddedKafkaBroker)
KafkaTestUtils.consumerProps(embeddedKafkaBroker, "grp", false)
);
ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
factory.setBatchListener(true);

2
spring-kafka/src/test/java/org/springframework/kafka/listener/SeekToCurrentRecovererTests.java
View File

@@ -89,7 +89,7 @@ public class SeekToCurrentRecovererTests {
@Test
public void testMaxFailures() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("seekTestMaxFailures", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "seekTestMaxFailures", false);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props, null,
new ErrorHandlingDeserializer<>(new JsonDeserializer<>(String.class)));

2
spring-kafka/src/test/java/org/springframework/kafka/listener/SubBatchPerPartitionTests.java
View File

@@ -134,7 +134,7 @@ public class SubBatchPerPartitionTests {
@Test
void defaults() {
Map<String, Object> props = KafkaTestUtils.consumerProps("sbpp", "false", this.broker);
Map<String, Object> props = KafkaTestUtils.consumerProps(this.broker, "sbpp", false);
ConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties("sbpp");
containerProps.setMessageListener(mock(MessageListener.class));

14
spring-kafka/src/test/java/org/springframework/kafka/listener/TransactionalContainerTests.java
View File

@@ -494,7 +494,7 @@ public class TransactionalContainerTests {
@SuppressWarnings({ "unchecked"})
@Test
public void testRollbackRecord() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("txTest1", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "txTest1", false);
props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -599,7 +599,7 @@ public class TransactionalContainerTests {
@SuppressWarnings({"unchecked"})
private void testFixLagGuts(String topic, int whichTm) throws InterruptedException {
Map<String, Object> props = KafkaTestUtils.consumerProps("txTest2", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "txTest2", false);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic);
@@ -655,7 +655,7 @@ public class TransactionalContainerTests {
@Test
public void testMaxFailures() throws Exception {
String group = "groupInARBP";
Map<String, Object> props = KafkaTestUtils.consumerProps(group, "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, group, false);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic3);
@@ -763,7 +763,7 @@ public class TransactionalContainerTests {
@Test
public void testBatchListenerMaxFailuresOnRecover() throws Exception {
String group = "groupInARBP2";
Map<String, Object> props = KafkaTestUtils.consumerProps(group, "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, group, false);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic8);
@@ -880,7 +880,7 @@ public class TransactionalContainerTests {
@SuppressWarnings("unchecked")
@Test
public void testRollbackProcessorCrash() throws Exception {
Map<String, Object> props = KafkaTestUtils.consumerProps("testRollbackNoRetries", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "testRollbackNoRetries", false);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic4);
@@ -941,7 +941,7 @@ public class TransactionalContainerTests {
@Test
public void testBatchListenerRecoverAfterRollbackProcessorCrash() throws Exception {
String group = "testBatchListenerRollbackNoRetries";
Map<String, Object> props = KafkaTestUtils.consumerProps(group, "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, group, false);
props.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 2);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
@@ -1095,7 +1095,7 @@ public class TransactionalContainerTests {
final KafkaTemplate<Object, Object> template = new KafkaTemplate<>(pf);
// init consumer
String group = "groupInARBP3";
Map<String, Object> consumerProperties = KafkaTestUtils.consumerProps(group, "false", embeddedKafka);
Map<String, Object> consumerProperties = KafkaTestUtils.consumerProps(embeddedKafka, group, false);
consumerProperties.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProperties);
ContainerProperties containerProps = new ContainerProperties(topic10);

8
spring-kafka/src/test/java/org/springframework/kafka/requestreply/ReplyingKafkaTemplateTests.java
View File

@@ -707,7 +707,7 @@ public class ReplyingKafkaTemplateTests {
}
});
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.testName, "false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, this.testName, false);
if (badDeser) {
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ErrorHandlingDeserializer.class);
consumerProps.put(ErrorHandlingDeserializer.VALUE_DESERIALIZER_CLASS, BadDeser.class);
@@ -731,7 +731,7 @@ public class ReplyingKafkaTemplateTests {
throws InterruptedException {
ContainerProperties containerProperties = new ContainerProperties(topic);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.testName, "false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, this.testName, false);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(consumerProps);
KafkaMessageListenerContainer<Integer, String> container = new KafkaMessageListenerContainer<>(cf,
containerProperties);
@@ -751,7 +751,7 @@ public class ReplyingKafkaTemplateTests {
ContainerProperties containerProperties = new ContainerProperties(topic);
containerProperties.setAckMode(AckMode.MANUAL_IMMEDIATE);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.testName, "false", embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(embeddedKafka, this.testName, false);
DefaultKafkaConsumerFactory<Integer, Collection<ConsumerRecord<Integer, String>>> cf =
new DefaultKafkaConsumerFactory<>(consumerProps);
KafkaMessageListenerContainer<Integer, Collection<ConsumerRecord<Integer, String>>> container =
@@ -887,7 +887,7 @@ public class ReplyingKafkaTemplateTests {
@Bean
public DefaultKafkaConsumerFactory<Integer, String> cf() {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("serverSide", "false", this.embeddedKafka);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.embeddedKafka, "serverSide", false);
return new DefaultKafkaConsumerFactory<>(consumerProps);
}

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicClassLevelIntegrationTests.java
View File

@@ -895,7 +895,7 @@ public class AsyncCompletableFutureRetryTopicClassLevelIntegrationTests {
Map<String, Object> props = KafkaTestUtils.consumerProps(
this.broker.getBrokersAsString(),
"groupId",
"false");
false);
props.put(
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncCompletableFutureRetryTopicScenarioTests.java
View File

@@ -1334,7 +1334,7 @@ public class AsyncCompletableFutureRetryTopicScenarioTests {
Map<String, Object> props = KafkaTestUtils.consumerProps(
this.broker.getBrokersAsString(),
"groupId",
"false");
false);
return new DefaultKafkaConsumerFactory<>(props);
}

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoFutureRetryTopicClassLevelIntegrationTests.java
View File

@@ -894,7 +894,7 @@ public class AsyncMonoFutureRetryTopicClassLevelIntegrationTests {
Map<String, Object> props = KafkaTestUtils.consumerProps(
this.broker.getBrokersAsString(),
"groupId",
"false");
false);
props.put(
ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class);

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/AsyncMonoRetryTopicScenarioTests.java
View File

@@ -1329,7 +1329,7 @@ public class AsyncMonoRetryTopicScenarioTests {
Map<String, Object> props = KafkaTestUtils.consumerProps(
this.broker.getBrokersAsString(),
"groupId",
"false");
false);
return new DefaultKafkaConsumerFactory<>(props);
}

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DeliveryHeaderTests.java
View File

@@ -157,7 +157,7 @@ public class DeliveryHeaderTests {
@Primary
ConsumerFactory<Integer, String> cf() {
return new DefaultKafkaConsumerFactory<>(
KafkaTestUtils.consumerProps("dh1", "false", this.broker));
KafkaTestUtils.consumerProps(this.broker, "dh1", false));
}
@Bean

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/DltStartupTests.java
View File

@@ -250,7 +250,7 @@ public class DltStartupTests {
@Bean
ConsumerFactory<Integer, String> cf(EmbeddedKafkaBroker broker) {
return new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps("dltStart", "false", broker));
return new DefaultKafkaConsumerFactory<>(KafkaTestUtils.consumerProps(broker, "dltStart", false));
}
@Bean

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/PartitionResolverTests.java
View File

@@ -115,7 +115,7 @@ public class PartitionResolverTests {
@Bean
ConsumerFactory<Integer, String> cf(EmbeddedKafkaBroker broker) {
Map<String, Object> props = KafkaTestUtils.consumerProps("prt", "false", broker);
Map<String, Object> props = KafkaTestUtils.consumerProps(broker, "prt", false);
return new DefaultKafkaConsumerFactory<>(props);
}

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationIntegrationTests.java
View File

@@ -126,7 +126,7 @@ class RetryTopicConfigurationIntegrationTests {
@Bean
ConsumerFactory<Integer, String> consumerFactory(EmbeddedKafkaBroker embeddedKafka) {
return new DefaultKafkaConsumerFactory<>(
KafkaTestUtils.consumerProps("retryConfig", "false", embeddedKafka));
KafkaTestUtils.consumerProps(embeddedKafka, "retryConfig", false));
}
@Bean

2
spring-kafka/src/test/java/org/springframework/kafka/retrytopic/RetryTopicConfigurationManualAssignmentIntegrationTests.java
View File

@@ -124,7 +124,7 @@ class RetryTopicConfigurationManualAssignmentIntegrationTests {
@Bean
ConsumerFactory<Integer, String> consumerFactory(EmbeddedKafkaBroker embeddedKafka) {
Map<String, Object> props = KafkaTestUtils.consumerProps("retryConfig", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps(embeddedKafka, "retryConfig", false);
props.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 2000);
return new DefaultKafkaConsumerFactory<>(
props);

2
spring-kafka/src/test/java/org/springframework/kafka/streams/KafkaStreamsBranchTests.java
View File

@@ -112,7 +112,7 @@ public class KafkaStreamsBranchTests {
private Consumer<String, String> createConsumer() {
Map<String, Object> consumerProps =
KafkaTestUtils.consumerProps(UUID.randomUUID().toString(), "false", this.embeddedKafka);
KafkaTestUtils.consumerProps(this.embeddedKafka, UUID.randomUUID().toString(), false);
consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10000);
DefaultKafkaConsumerFactory<String, String> kafkaConsumerFactory =

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

@@ -226,8 +226,7 @@ class KafkaStreamsInteractiveQueryServiceTests {
@Bean
public Map<String, Object> consumerConfigs() {
return KafkaTestUtils.consumerProps(this.brokerAddresses, "testGroup",
"false");
return KafkaTestUtils.consumerProps(this.brokerAddresses, "testGroup", false);
}
@Bean

2
spring-kafka/src/test/java/org/springframework/kafka/streams/KafkaStreamsJsonSerializationTests.java
View File

@@ -117,7 +117,7 @@ public class KafkaStreamsJsonSerializationTests {
private <K, V> Consumer<K, V> consumer(String topic, Serde<K> keySerde, Serde<V> valueSerde) {
Map<String, Object> consumerProps =
KafkaTestUtils.consumerProps(UUID.randomUUID().toString(), "false", this.embeddedKafka);
KafkaTestUtils.consumerProps(this.embeddedKafka, UUID.randomUUID().toString(), false);
consumerProps.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 10000);
DefaultKafkaConsumerFactory<K, V> kafkaConsumerFactory =

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

@@ -243,9 +243,7 @@ public class KafkaStreamsTests {
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.brokerAddresses, "testGroup",
"false");
return consumerProps;
return KafkaTestUtils.consumerProps(this.brokerAddresses, "testGroup", false);
}
@Bean

3
spring-kafka/src/test/java/org/springframework/kafka/streams/RecoveringDeserializationExceptionHandlerTests.java
View File

@@ -215,8 +215,7 @@ public class RecoveringDeserializationExceptionHandlerTests {
@Bean
public Map<String, Object> consumerConfigs() {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.brokerAddresses, "recovererGroup",
"false");
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(this.brokerAddresses, "recovererGroup", false);
consumerProps.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
consumerProps.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, ByteArrayDeserializer.class);
return consumerProps;

2
spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/MicrometerMetricsTests.java
View File

@@ -111,7 +111,7 @@ public class MicrometerMetricsTests {
@Bean
ConsumerFactory<Integer, String> consumerFactory(EmbeddedKafkaBroker broker) {
return new DefaultKafkaConsumerFactory<>(
KafkaTestUtils.consumerProps("metrics", "false", broker));
KafkaTestUtils.consumerProps(broker, "metrics", false));
}
@Bean

2
spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationIntegrationTests.java
View File

@@ -131,7 +131,7 @@ public class ObservationIntegrationTests extends SampleTestRunner {
@Bean
ConsumerFactory<Integer, String> consumerFactory(EmbeddedKafkaBroker broker) {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("obs", "false", broker);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(broker, "obs", false);
return new DefaultKafkaConsumerFactory<>(consumerProps);
}

2
spring-kafka/src/test/java/org/springframework/kafka/support/micrometer/ObservationTests.java
View File

@@ -570,7 +570,7 @@ public class ObservationTests {
@Bean
ConsumerFactory<Integer, String> consumerFactory(EmbeddedKafkaBroker broker) {
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps("obs", "false", broker);
Map<String, Object> consumerProps = KafkaTestUtils.consumerProps(broker, "obs", false);
consumerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, broker.getBrokersAsString() + ","
+ broker.getBrokersAsString() + "," + broker.getBrokersAsString());
return new DefaultKafkaConsumerFactory<>(consumerProps);

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

@@ -48,7 +48,7 @@ public class SerializationIntegrationTests {
@Test
void configurePreLoadedDelegates() {
Map<String, Object> consumerProps =
KafkaTestUtils.consumerProps(DBTD_TOPIC, "false", EmbeddedKafkaCondition.getBroker());
KafkaTestUtils.consumerProps(EmbeddedKafkaCondition.getBroker(), DBTD_TOPIC, false);
consumerProps.put(DelegatingByTopicDeserializer.VALUE_SERIALIZATION_TOPIC_CONFIG, DBTD_TOPIC + ":"
+ TestDeserializer.class.getName());
TestDeserializer testDeser = new TestDeserializer();