Skip to content

Commit 637738b

Browse files
garyrussellartembilan
garyrussell
authored and
artembilan
committed
GH-945: enable.auto.commit=false by default
Resolves #945 Turn off auto commit by default.
1 parent 1eead9d commit 637738b

17 files changed

+138
-38
lines changed

‎spring-kafka-test/src/main/java/org/springframework/kafka/test/utils/KafkaTestUtils.java‎

Lines changed: 18 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2016-2018 the original author or authors.
2+
* Copyright 2016-2019 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -21,6 +21,7 @@
2121
import java.time.Duration;
2222
import java.util.HashMap;
2323
import java.util.Map;
24+
import java.util.Properties;
2425
import java.util.stream.Collectors;
2526

2627
import org.apache.commons.logging.Log;
@@ -50,6 +51,8 @@ public final class KafkaTestUtils {
5051

5152
private static final Log logger = LogFactory.getLog(KafkaTestUtils.class); // NOSONAR
5253

54+
private static Properties defaults;
55+
5356
private KafkaTestUtils() {
5457
// private ctor
5558
}
@@ -249,4 +252,18 @@ public static <T> T getPropertyValue(Object root, String propertyPath, Class<T>
249252
return (T) value;
250253
}
251254

255+
/**
256+
* Return a {@link Properties} object equal to the default consumer property overrides.
257+
* Useful when matching arguments in Mockito tests.
258+
* @return the default properties.
259+
* @since 2.2.5
260+
*/
261+
public static Properties defaultPropertyOverrides() {
262+
if (defaults == null) {
263+
Properties props = new Properties();
264+
props.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
265+
defaults = props;
266+
}
267+
return defaults;
268+
}
252269
}

‎spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java‎

Lines changed: 26 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import java.util.List;
3131
import java.util.Map;
3232
import java.util.Map.Entry;
33+
import java.util.Properties;
3334
import java.util.Set;
3435
import java.util.concurrent.BlockingQueue;
3536
import java.util.concurrent.ConcurrentHashMap;
@@ -417,7 +418,7 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
417418

418419
private final boolean wantsFullRecords;
419420

420-
private final boolean autoCommit = KafkaMessageListenerContainer.this.consumerFactory.isAutoCommit();
421+
private final boolean autoCommit;
421422

422423
private final boolean isManualAck = this.containerProperties.getAckMode().equals(AckMode.MANUAL);
423424

@@ -493,15 +494,14 @@ private final class ListenerConsumer implements SchedulingAwareRunnable, Consume
493494

494495
@SuppressWarnings(UNCHECKED)
495496
ListenerConsumer(GenericMessageListener<?> listener, ListenerType listenerType) {
496-
Assert.state(!this.isAnyManualAck || !this.autoCommit,
497-
() -> "Consumer cannot be configured for auto commit for ackMode "
498-
+ this.containerProperties.getAckMode());
497+
Properties consumerProperties = new Properties(this.containerProperties.getConsumerProperties());
498+
this.autoCommit = determineAutoCommit(consumerProperties);
499499
this.consumer =
500500
KafkaMessageListenerContainer.this.consumerFactory.createConsumer(
501501
this.consumerGroupId,
502502
this.containerProperties.getClientId(),
503503
KafkaMessageListenerContainer.this.clientIdSuffix,
504-
this.containerProperties.getConsumerProperties());
504+
consumerProperties);
505505

506506
this.transactionTemplate = determineTransactionTemplate();
507507
subscribeOrAssignTopics(this.consumer);
@@ -572,6 +572,27 @@ private TransactionTemplate determineTransactionTemplate() {
572572
: null;
573573
}
574574

575+
private boolean determineAutoCommit(Properties consumerProperties) {
576+
boolean autoCommit;
577+
String autoCommitOverride = consumerProperties.getProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
578+
if (!KafkaMessageListenerContainer.this.consumerFactory.getConfigurationProperties()
579+
.containsKey(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)
580+
&& autoCommitOverride == null) {
581+
consumerProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");
582+
autoCommit = false;
583+
}
584+
else if (autoCommitOverride != null) {
585+
autoCommit = Boolean.parseBoolean(autoCommitOverride);
586+
}
587+
else {
588+
autoCommit = KafkaMessageListenerContainer.this.consumerFactory.isAutoCommit();
589+
}
590+
Assert.state(!this.isAnyManualAck || !autoCommit,
591+
() -> "Consumer cannot be configured for auto commit for ackMode "
592+
+ this.containerProperties.getAckMode());
593+
return autoCommit;
594+
}
595+
575596
private Duration determineSyncCommitTimeout() {
576597
if (this.containerProperties.getSyncCommitTimeout() != null) {
577598
return this.containerProperties.getSyncCommitTimeout();

‎spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerMockTests.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,6 @@
2323
import static org.mockito.Mockito.mock;
2424

2525
import java.util.Collections;
26-
import java.util.Properties;
2726
import java.util.concurrent.CountDownLatch;
2827
import java.util.concurrent.TimeUnit;
2928
import java.util.concurrent.atomic.AtomicBoolean;
@@ -34,6 +33,7 @@
3433
import org.junit.jupiter.api.Test;
3534

3635
import org.springframework.kafka.core.ConsumerFactory;
36+
import org.springframework.kafka.test.utils.KafkaTestUtils;
3737

3838
/**
3939
* @author Gary Russell
@@ -55,7 +55,8 @@ public void testCorrectContainerForConsumerError() throws InterruptedException {
5555
Thread.sleep(100);
5656
return new ConsumerRecords<>(Collections.emptyMap());
5757
}).given(consumer).poll(any());
58-
given(consumerFactory.createConsumer("grp", "", "-0", new Properties())).willReturn(consumer);
58+
given(consumerFactory.createConsumer("grp", "", "-0", KafkaTestUtils.defaultPropertyOverrides()))
59+
.willReturn(consumer);
5960
ContainerProperties containerProperties = new ContainerProperties("foo");
6061
containerProperties.setGroupId("grp");
6162
containerProperties.setMessageListener((MessageListener) record -> { });

‎spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java‎

Lines changed: 50 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -30,6 +30,7 @@
3030
import java.util.HashSet;
3131
import java.util.List;
3232
import java.util.Map;
33+
import java.util.Properties;
3334
import java.util.Set;
3435
import java.util.concurrent.ConcurrentSkipListSet;
3536
import java.util.concurrent.CountDownLatch;
@@ -43,6 +44,7 @@
4344
import org.apache.kafka.clients.consumer.ConsumerConfig;
4445
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
4546
import org.apache.kafka.clients.consumer.ConsumerRecords;
47+
import org.apache.kafka.clients.consumer.KafkaConsumer;
4648
import org.apache.kafka.common.TopicPartition;
4749
import org.junit.ClassRule;
4850
import org.junit.Test;
@@ -105,7 +107,18 @@ public class ConcurrentMessageListenerContainerTests {
105107
public void testAutoCommit() throws Exception {
106108
this.logger.info("Start auto");
107109
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka);
108-
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
110+
AtomicReference<Properties> overrides = new AtomicReference<>();
111+
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {
112+
113+
@Override
114+
protected KafkaConsumer<Integer, String> createKafkaConsumer(String groupId, String clientIdPrefix,
115+
String clientIdSuffixArg, Properties properties) {
116+
117+
overrides.set(properties);
118+
return super.createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, properties);
119+
}
120+
121+
};
109122
ContainerProperties containerProps = new ContainerProperties(topic1);
110123
containerProps.setLogContainerConfig(true);
111124

@@ -172,14 +185,26 @@ public void testAutoCommit() throws Exception {
172185
assertThat(children).contains((KafkaMessageListenerContainer<Integer, String>) e.getSource());
173186
}
174187
});
188+
assertThat(overrides.get().getProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)).isNull();
175189
this.logger.info("Stop auto");
176190
}
177191

178192
@Test
179193
public void testAutoCommitWithRebalanceListener() throws Exception {
180194
this.logger.info("Start auto");
181-
Map<String, Object> props = KafkaTestUtils.consumerProps("test10", "true", embeddedKafka);
182-
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
195+
Map<String, Object> props = KafkaTestUtils.consumerProps("test10", "false", embeddedKafka);
196+
AtomicReference<Properties> overrides = new AtomicReference<>();
197+
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {
198+
199+
@Override
200+
protected KafkaConsumer<Integer, String> createKafkaConsumer(String groupId, String clientIdPrefix,
201+
String clientIdSuffixArg, Properties properties) {
202+
203+
overrides.set(properties);
204+
return super.createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, properties);
205+
}
206+
207+
};
183208
ContainerProperties containerProps = new ContainerProperties(topic1);
184209

185210
final CountDownLatch latch = new CountDownLatch(4);
@@ -189,6 +214,9 @@ public void testAutoCommitWithRebalanceListener() throws Exception {
189214
listenerThreadNames.add(Thread.currentThread().getName());
190215
latch.countDown();
191216
});
217+
Properties consumerProperties = new Properties();
218+
consumerProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
219+
containerProps.setConsumerProperties(consumerProperties);
192220
final CountDownLatch rebalancePartitionsAssignedLatch = new CountDownLatch(2);
193221
final CountDownLatch rebalancePartitionsRevokedLatch = new CountDownLatch(2);
194222
containerProps.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
@@ -231,14 +259,27 @@ public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
231259
assertThat(threadName).contains("-C-");
232260
}
233261
container.stop();
262+
assertThat(overrides.get().getProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)).isEqualTo("true");
234263
this.logger.info("Stop auto");
235264
}
236265

237266
@Test
238267
public void testAfterListenCommit() throws Exception {
239268
this.logger.info("Start manual");
240269
Map<String, Object> props = KafkaTestUtils.consumerProps("test2", "false", embeddedKafka);
241-
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
270+
props.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
271+
AtomicReference<Properties> overrides = new AtomicReference<>();
272+
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props) {
273+
274+
@Override
275+
protected KafkaConsumer<Integer, String> createKafkaConsumer(String groupId, String clientIdPrefix,
276+
String clientIdSuffixArg, Properties properties) {
277+
278+
overrides.set(properties);
279+
return super.createKafkaConsumer(groupId, clientIdPrefix, clientIdSuffixArg, properties);
280+
}
281+
282+
};
242283
ContainerProperties containerProps = new ContainerProperties(topic2);
243284

244285
final CountDownLatch latch = new CountDownLatch(4);
@@ -267,6 +308,7 @@ public void testAfterListenCommit() throws Exception {
267308
assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
268309
container.stop();
269310
this.logger.info("Stop manual");
311+
assertThat(overrides.get().getProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)).isEqualTo("false");
270312
}
271313

272314
@Test
@@ -456,6 +498,7 @@ public ConsumerRecords<Integer, String> answer(InvocationOnMock invocation) thro
456498
public void testListenerException() throws Exception {
457499
this.logger.info("Start exception");
458500
Map<String, Object> props = KafkaTestUtils.consumerProps("test1", "true", embeddedKafka);
501+
props.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG);
459502
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
460503
ContainerProperties containerProps = new ContainerProperties(topic6);
461504
containerProps.setAckCount(23);
@@ -467,6 +510,9 @@ public void testListenerException() throws Exception {
467510
latch.countDown();
468511
throw new RuntimeException("intended");
469512
});
513+
Properties consumerProperties = new Properties();
514+
consumerProperties.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
515+
containerProps.setConsumerProperties(consumerProperties);
470516

471517
ConcurrentMessageListenerContainer<Integer, String> container =
472518
new ConcurrentMessageListenerContainer<>(cf, containerProps);

‎spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingBatchErrorHandlerTests.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232
import java.util.LinkedHashMap;
3333
import java.util.List;
3434
import java.util.Map;
35-
import java.util.Properties;
3635
import java.util.concurrent.CountDownLatch;
3736
import java.util.concurrent.TimeUnit;
3837
import java.util.concurrent.atomic.AtomicInteger;
@@ -55,6 +54,7 @@
5554
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
5655
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
5756
import org.springframework.kafka.core.ConsumerFactory;
57+
import org.springframework.kafka.test.utils.KafkaTestUtils;
5858
import org.springframework.test.annotation.DirtiesContext;
5959
import org.springframework.test.context.junit4.SpringRunner;
6060

@@ -136,7 +136,8 @@ public void foo(List<String> in) {
136136
public ConsumerFactory consumerFactory() {
137137
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
138138
final Consumer consumer = consumer();
139-
given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0", new Properties())).willReturn(consumer);
139+
given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0", KafkaTestUtils.defaultPropertyOverrides()))
140+
.willReturn(consumer);
140141
return consumerFactory;
141142
}
142143

‎spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerBatchModeTests.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@
3131
import java.util.LinkedHashMap;
3232
import java.util.List;
3333
import java.util.Map;
34-
import java.util.Properties;
3534
import java.util.concurrent.CountDownLatch;
3635
import java.util.concurrent.TimeUnit;
3736
import java.util.concurrent.atomic.AtomicInteger;
@@ -54,6 +53,7 @@
5453
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
5554
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
5655
import org.springframework.kafka.core.ConsumerFactory;
56+
import org.springframework.kafka.test.utils.KafkaTestUtils;
5757
import org.springframework.test.annotation.DirtiesContext;
5858
import org.springframework.test.context.junit4.SpringRunner;
5959

@@ -133,7 +133,8 @@ public void foo(String in) {
133133
public ConsumerFactory consumerFactory() {
134134
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
135135
final Consumer consumer = consumer();
136-
given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0", new Properties())).willReturn(consumer);
136+
given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0", KafkaTestUtils.defaultPropertyOverrides()))
137+
.willReturn(consumer);
137138
return consumerFactory;
138139
}
139140

‎spring-kafka/src/test/java/org/springframework/kafka/listener/ContainerStoppingErrorHandlerRecordModeTests.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232
import java.util.LinkedHashMap;
3333
import java.util.List;
3434
import java.util.Map;
35-
import java.util.Properties;
3635
import java.util.concurrent.CountDownLatch;
3736
import java.util.concurrent.TimeUnit;
3837
import java.util.concurrent.atomic.AtomicInteger;
@@ -56,6 +55,7 @@
5655
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
5756
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
5857
import org.springframework.kafka.core.ConsumerFactory;
58+
import org.springframework.kafka.test.utils.KafkaTestUtils;
5959
import org.springframework.test.annotation.DirtiesContext;
6060
import org.springframework.test.context.junit4.SpringRunner;
6161

@@ -147,7 +147,8 @@ public void foo(String in) {
147147
public ConsumerFactory consumerFactory() {
148148
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
149149
final Consumer consumer = consumer();
150-
given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0", new Properties())).willReturn(consumer);
150+
given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0", KafkaTestUtils.defaultPropertyOverrides()))
151+
.willReturn(consumer);
151152
return consumerFactory;
152153
}
153154

‎spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1028,8 +1028,8 @@ public void testSeekAutoCommit() throws Exception {
10281028
@Test
10291029
public void testSeekAutoCommitDefault() throws Exception {
10301030
Map<String, Object> props = KafkaTestUtils.consumerProps("test15", "true", embeddedKafka);
1031-
props.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); // test true by default
1032-
testSeekGuts(props, topic15, true);
1031+
props.remove(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG); // test false by default
1032+
testSeekGuts(props, topic15, false);
10331033
}
10341034

10351035
@Test

‎spring-kafka/src/test/java/org/springframework/kafka/listener/RemainingRecordsErrorHandlerTests.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,6 @@
3232
import java.util.LinkedHashMap;
3333
import java.util.List;
3434
import java.util.Map;
35-
import java.util.Properties;
3635
import java.util.concurrent.CountDownLatch;
3736
import java.util.concurrent.TimeUnit;
3837
import java.util.concurrent.atomic.AtomicInteger;
@@ -56,6 +55,7 @@
5655
import org.springframework.kafka.annotation.KafkaListener;
5756
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
5857
import org.springframework.kafka.core.ConsumerFactory;
58+
import org.springframework.kafka.test.utils.KafkaTestUtils;
5959
import org.springframework.test.annotation.DirtiesContext;
6060
import org.springframework.test.context.junit4.SpringRunner;
6161

@@ -137,7 +137,8 @@ public void foo(String in) {
137137
public ConsumerFactory consumerFactory() {
138138
ConsumerFactory consumerFactory = mock(ConsumerFactory.class);
139139
final Consumer consumer = consumer();
140-
given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0", new Properties())).willReturn(consumer);
140+
given(consumerFactory.createConsumer(CONTAINER_ID, "", "-0", KafkaTestUtils.defaultPropertyOverrides()))
141+
.willReturn(consumer);
141142
return consumerFactory;
142143
}
143144

0 commit comments

Comments
 (0)