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

GH-62; commitSync() By Default

Browse Source 
Fixes #62
Resolves #72

See the discussion on GH-62 `commitAsync()` is not currenly reliable.
Use `commitSync()` by default; add `syncCommits` property to the containers
(default true).

Also allow a user-injected commit callback (GH-72)
This commit is contained in:
Gary Russell
2016-05-12 15:37:06 -04:00
parent 964beb67e2
commit d93d91b339
3 changed files with 94 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
spring-kafka/src/main/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainer.java
View File

@@ -24,6 +24,7 @@ import java.util.List;
import java.util.regex.Pattern;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import org.springframework.kafka.core.ConsumerFactory;
@@ -61,6 +62,10 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
private ConsumerRebalanceListener consumerRebalanceListener;
private OffsetCommitCallback commitCallback;
private boolean syncCommits;
/**
* Construct an instance with the supplied configuration properties and specific
* topics/partitions - when using this constructor, {@link #setRecentOffset(long)
@@ -146,6 +151,26 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
this.consumerRebalanceListener = consumerRebalanceListener;
}
/**
* Set the commit callback; by default a simple logging callback is used to
* log success at DEBUG level and failures at ERROR level.
* @param commitCallback the callback.
*/
public void setCommitCallback(OffsetCommitCallback commitCallback) {
this.commitCallback = commitCallback;
}
/**
* Set whether or not to call consumer.commitSync() or commitAsync() when
* the container is responsible for commits. Default true. See
* https://github.com/spring-projects/spring-kafka/issues/62
* At the time of writing, async commits are not entirely reliable.
* @param syncCommits true to use commitSync().
*/
public void setSyncCommits(boolean syncCommits) {
this.syncCommits = syncCommits;
}
/**
* Return the list of {@link KafkaMessageListenerContainer}s created by
* this container.
@@ -180,6 +205,8 @@ public class ConcurrentMessageListenerContainer<K, V> extends AbstractMessageLis
container = new KafkaMessageListenerContainer<>(this.consumerFactory, this.consumerRebalanceListener,
this.topics, this.topicPattern, partitionSubset(i));
}
container.setCommitCallback(this.commitCallback);
container.setSyncCommits(this.syncCommits);
container.setAckMode(getAckMode());
container.setAckCount(getAckCount());
container.setAckTime(getAckTime());

55
spring-kafka/src/main/java/org/springframework/kafka/listener/KafkaMessageListenerContainer.java
View File

@@ -68,6 +68,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
private final TopicPartition[] partitions;
private final ConsumerRebalanceListener consumerRebalanceListener;
private ListenerConsumer listenerConsumer;
private long recentOffset;
@@ -76,7 +78,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
private AcknowledgingMessageListener<K, V> acknowledgingMessageListener;
private final ConsumerRebalanceListener consumerRebalanceListener;
private OffsetCommitCallback commitCallback;
private boolean syncCommits = true;
/**
* Construct an instance with the supplied configuration properties and specific
@@ -195,6 +199,25 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
this.recentOffset = recentOffset;
}
/**
* Set the commit callback; by default a simple logging callback is used to
* log success at DEBUG level and failures at ERROR level.
* @param commitCallback the callback.
*/
public void setCommitCallback(OffsetCommitCallback commitCallback) {
this.commitCallback = commitCallback;
}
/**
* Set whether or not to call consumer.commitSync() or commitAsync() when
* the container is responsible for commits. Default true. See
* https://github.com/spring-projects/spring-kafka/issues/62
* At the time of writing, async commits are not entirely reliable.
* @param syncCommits true to use commitSync().
*/
public void setSyncCommits(boolean syncCommits) {
this.syncCommits = syncCommits;
}
/**
* Return the {@link TopicPartition}s currently assigned to this container,
@@ -275,7 +298,9 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
private final Log logger = LogFactory.getLog(ListenerConsumer.class);
private final CommitCallback callback = new CommitCallback();
private final OffsetCommitCallback commitCallback = KafkaMessageListenerContainer.this.commitCallback != null
? KafkaMessageListenerContainer.this.commitCallback
: new LoggingCommitCallback();
private final Consumer<K, V> consumer;
@@ -293,6 +318,8 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
private final AckMode ackMode = getAckMode();
private final boolean syncCommits = KafkaMessageListenerContainer.this.syncCommits;
private Thread consumerThread;
private volatile Collection<TopicPartition> definedPartitions;
@@ -376,7 +403,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
if (!this.autoCommit && this.ackMode.equals(AckMode.RECORD)) {
this.consumer.commitAsync(
Collections.singletonMap(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset() + 1)), this.callback);
new OffsetAndMetadata(record.offset() + 1)), this.commitCallback);
}
}
if (!this.autoCommit) {
@@ -446,7 +473,7 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
}
if (ListenerConsumer.this.ackMode.equals(AckMode.MANUAL_IMMEDIATE)) {
ListenerConsumer.this.consumer.commitAsync(commits,
ListenerConsumer.this.callback);
ListenerConsumer.this.commitCallback);
}
else {
ListenerConsumer.this.consumer.commitSync(commits);
@@ -485,10 +512,15 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
long now;
if (ackMode.equals(AckMode.BATCH)) {
if (!records.isEmpty()) {
this.consumer.commitAsync(this.callback);
if (this.syncCommits) {
this.consumer.commitSync();
}
else {
this.consumer.commitAsync(this.commitCallback);
}
}
}
else if (!ackMode.equals(AckMode.MANUAL_IMMEDIATE)) {
else if (!ackMode.equals(AckMode.MANUAL_IMMEDIATE) && !ackMode.equals(AckMode.MANUAL_IMMEDIATE_SYNC)) {
if (!ackMode.equals(AckMode.MANUAL)) {
updatePendingOffsets(records);
}
@@ -575,14 +607,19 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
this.logger.debug("Committing: " + commits);
}
if (!commits.isEmpty()) {
this.consumer.commitAsync(commits, this.callback);
if (this.syncCommits) {
this.consumer.commitSync(commits);
}
else {
this.consumer.commitAsync(commits, this.commitCallback);
}
}
}
}
private static final class CommitCallback implements OffsetCommitCallback {
private static final class LoggingCommitCallback implements OffsetCommitCallback {
private static final Log logger = LogFactory.getLog(OffsetCommitCallback.class);
private static final Log logger = LogFactory.getLog(LoggingCommitCallback.class);
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {

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

@@ -36,11 +36,12 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.common.TopicPartition;
import org.junit.ClassRule;
import org.junit.Ignore;
import org.junit.Test;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
@@ -332,10 +333,10 @@ public class ConcurrentMessageListenerContainerTests {
@Test
public void testManualCommit() throws Exception {
testManualCommitGuts(AckMode.MANUAL, topic4);
testManualCommitGuts(AckMode.MANUAL_IMMEDIATE, topic5);
testManualCommitGuts(AckMode.MANUAL_IMMEDIATE_SYNC, topic5);
// to be sure the commits worked ok so run the tests again and the second tests start at the committed offset.
testManualCommitGuts(AckMode.MANUAL, topic4);
testManualCommitGuts(AckMode.MANUAL_IMMEDIATE, topic5);
testManualCommitGuts(AckMode.MANUAL_IMMEDIATE_SYNC, topic5);
}
private void testManualCommitGuts(AckMode ackMode, String topic) throws Exception {
@@ -375,6 +376,7 @@ public class ConcurrentMessageListenerContainerTests {
}
@Test
@Ignore // TODO https://github.com/spring-projects/spring-kafka/issues/62 using SYNC for avoidance
public void testManualCommitExisting() throws Exception {
logger.info("Start MANUAL_IMMEDIATE with Existing");
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
@@ -405,6 +407,19 @@ public class ConcurrentMessageListenerContainerTests {
container.setConcurrency(1);
container.setAckMode(AckMode.MANUAL_IMMEDIATE);
container.setBeanName("testManualExisting");
final CountDownLatch commits = new CountDownLatch(8);
final AtomicReference<Exception> exceptionRef = new AtomicReference<Exception>();
container.setCommitCallback(new OffsetCommitCallback() {
@Override
public void onComplete(Map<TopicPartition, OffsetAndMetadata> offsets, Exception exception) {
commits.countDown();
if (exception != null) {
exceptionRef.compareAndSet(null, exception);
}
}
});
container.start();
ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
template.send(0, "fooo");
@@ -413,6 +428,8 @@ public class ConcurrentMessageListenerContainerTests {
template.send(2, "quxx");
template.flush();
assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(commits.await(60, TimeUnit.SECONDS)).isTrue();
assertThat(exceptionRef.get()).isNull();
container.stop();
logger.info("Stop MANUAL_IMMEDIATE with Existing");
}