diff --git a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java index 923248d7..33663396 100644 --- a/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java +++ b/spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java @@ -172,7 +172,7 @@ public class ContainerProperties extends ConsumerProperties { private boolean logContainerConfig; - private boolean missingTopicsFatal = true; + private boolean missingTopicsFatal = false; private long idleBetweenPolls; @@ -482,7 +482,7 @@ public class ContainerProperties extends ConsumerProperties { /** * If true, the container won't start if any of the configured topics are not present - * on the broker. Does not apply when topic patterns are configured. Default true; + * on the broker. Does not apply when topic patterns are configured. Default false. * @return the missingTopicsFatal. * @since 2.2 */ diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java index 451d4507..5deba67f 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java @@ -31,7 +31,6 @@ import static org.mockito.Mockito.never; import static org.mockito.Mockito.spy; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.withSettings; import java.time.Duration; import java.util.ArrayList; @@ -2269,7 +2268,7 @@ public class KafkaMessageListenerContainerTests { @Test public void testPauseResumeAndConsumerSeekAware() throws Exception { ConsumerFactory cf = mock(ConsumerFactory.class); - Consumer consumer = mock(Consumer.class, withSettings().verboseLogging()); + Consumer consumer = mock(Consumer.class); given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer); Map cfProps = new LinkedHashMap<>(); cfProps.put(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG, 45000); @@ -2294,11 +2293,11 @@ public class KafkaMessageListenerContainerTests { } return first.getAndSet(false) ? consumerRecords : emptyRecords; }); - final CountDownLatch commitLatch = new CountDownLatch(5); // assignment + 4 + final CountDownLatch seekLatch = new CountDownLatch(7); willAnswer(i -> { - commitLatch.countDown(); + seekLatch.countDown(); return null; - }).given(consumer).commitSync(anyMap(), any()); + }).given(consumer).seekToEnd(any()); given(consumer.assignment()).willReturn(records.keySet()); final CountDownLatch pauseLatch1 = new CountDownLatch(2); // consumer, event publisher final CountDownLatch pauseLatch2 = new CountDownLatch(2); // consumer, consumer @@ -2371,7 +2370,7 @@ public class KafkaMessageListenerContainerTests { } }); container.start(); - assertThat(commitLatch.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(seekLatch.await(10, TimeUnit.SECONDS)).isTrue(); InOrder inOrder = inOrder(consumer); inOrder.verify(consumer).commitSync(anyMap(), eq(Duration.ofSeconds(41))); diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingTopicsTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingTopicsTests.java index 67796800..264cf30a 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingTopicsTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/MissingTopicsTests.java @@ -51,9 +51,11 @@ public class MissingTopicsTests { DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties("notexisting"); containerProps.setMessageListener((MessageListener) message -> { }); + containerProps.setMissingTopicsFatal(true); ConcurrentMessageListenerContainer container = new ConcurrentMessageListenerContainer<>(cf, containerProps); container.setBeanName("testMissing1"); + try { container.start(); fail("Expected exception"); @@ -69,6 +71,7 @@ public class MissingTopicsTests { DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); ContainerProperties containerProps = new ContainerProperties("notexisting"); containerProps.setMessageListener((MessageListener) message -> { }); + containerProps.setMissingTopicsFatal(true); KafkaMessageListenerContainer container = new KafkaMessageListenerContainer<>(cf, containerProps); container.setBeanName("testMissing2"); diff --git a/src/reference/asciidoc/whats-new.adoc b/src/reference/asciidoc/whats-new.adoc index a428bd93..c796f1cf 100644 --- a/src/reference/asciidoc/whats-new.adoc +++ b/src/reference/asciidoc/whats-new.adoc @@ -18,6 +18,11 @@ This version requires the 2.3.0 `kafka-clients` or higher. `TopicPartitionInitialOffset` is deprecated in favor of `TopicPartitionOffset`. +==== Configuration Changes + +Starting with version 2.3.4, the `missingTopicsFatal` container property is false by default. +When this is true, the application fails to start if the broker is down; many users were affected by this change; given that Kafka is a high-availability platform, we did not anticipate that starting an application with no active brokers would be a common use case. + ==== Producer and Consumer Factory Changes The `DefaultKafkaProducerFactory` can now be configured to create a producer per thread.