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

missingTopicsFatal false by default

Browse Source 
* Fix tests

* Fix race in KMLCT.testPauseResumeAndConsumerSeekAware().
This commit is contained in:
Gary Russell
2019-11-14 13:45:37 -05:00
committed by Artem Bilan
parent c91d491d10
commit 29403eb230
4 changed files with 15 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
spring-kafka/src/main/java/org/springframework/kafka/listener/ContainerProperties.java
View File

@@ -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
*/

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

@@ -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<Integer, String> cf = mock(ConsumerFactory.class);
Consumer<Integer, String> consumer = mock(Consumer.class, withSettings().verboseLogging());
Consumer<Integer, String> consumer = mock(Consumer.class);
given(cf.createConsumer(eq("grp"), eq("clientId"), isNull(), any())).willReturn(consumer);
Map<String, Object> 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)));

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

@@ -51,9 +51,11 @@ public class MissingTopicsTests {
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties("notexisting");
containerProps.setMessageListener((MessageListener<Integer, String>) message -> { });
containerProps.setMissingTopicsFatal(true);
ConcurrentMessageListenerContainer<Integer, String> container =
new ConcurrentMessageListenerContainer<>(cf, containerProps);
container.setBeanName("testMissing1");
try {
container.start();
fail("Expected exception");
@@ -69,6 +71,7 @@ public class MissingTopicsTests {
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties("notexisting");
containerProps.setMessageListener((MessageListener<Integer, String>) message -> { });
containerProps.setMissingTopicsFatal(true);
KafkaMessageListenerContainer<Integer, String> container =
new KafkaMessageListenerContainer<>(cf, containerProps);
container.setBeanName("testMissing2");

5
src/reference/asciidoc/whats-new.adoc
View File

@@ -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.