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

GH-511: Committing offset immediately on error

Browse Source 
Fixes: spring-projects/spring-kafka#511

**Cherry-pick to 2.0.x & master**

# Conflicts:
#	spring-kafka/src/test/java/org/springframework/kafka/listener/KafkaMessageListenerContainerTests.java
This commit is contained in:
Tom van den Berge
2017-12-11 16:44:07 +01:00
committed by Artem Bilan
parent 36aabb40af
commit ebf7bb1d73
2 changed files with 121 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -88,6 +88,7 @@ import org.springframework.util.concurrent.ListenableFutureCallback;
* @author Loic Talhouarne
* @author Vladimir Tsanev
* @author Yang Qiju
* @author Tom van den Berge
*/
public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListenerContainer<K, V> {
@@ -1003,7 +1004,20 @@ public class KafkaMessageListenerContainer<K, V> extends AbstractMessageListener
}
catch (RuntimeException e) {
if (this.containerProperties.isAckOnError() && !this.autoCommit && producer == null) {
this.acks.add(record);
if (this.isRecordAck) {
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit =
Collections.singletonMap(new TopicPartition(record.topic(), record.partition()),
new OffsetAndMetadata(record.offset() + 1));
if (this.containerProperties.isSyncCommits()) {
this.consumer.commitSync(offsetsToCommit);
}
else {
this.consumer.commitAsync(offsetsToCommit, this.commitCallback);
}
}
else if (!this.isAnyManualAck) {
this.acks.add(record);
}
}
if (this.errorHandler == null) {
throw e;

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

@@ -89,6 +89,7 @@ import org.springframework.kafka.test.utils.KafkaTestUtils;
* @author Martin Dam
* @author Artem Bilan
* @author Loic Talhouarne
* @author Tom van den Berge
*/
public class KafkaMessageListenerContainerTests {
@@ -128,10 +129,14 @@ public class KafkaMessageListenerContainerTests {
private static String topic17 = "testTopic17";
private static String topic18 = "testTopic18";
private static String topic19 = "testTopic19";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5,
topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16,
topic17);
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic5,
topic6, topic7, topic8, topic9, topic10, topic11, topic12, topic13, topic14, topic15, topic16, topic17, topic18,
topic19);
@Rule
public TestName testName = new TestName();
@@ -1518,6 +1523,101 @@ public class KafkaMessageListenerContainerTests {
this.logger.info("Stop JSON2");
}
@Test
public void testRebalanceAfterFailedRecord() throws Exception {
logger.info("Start rebalance after failed record");
Map<String, Object> props = KafkaTestUtils.consumerProps("test18", "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<>(props);
ContainerProperties containerProps = new ContainerProperties(topic18);
final List<AtomicInteger> counts = new ArrayList<>();
counts.add(new AtomicInteger());
counts.add(new AtomicInteger());
containerProps.setMessageListener(new MessageListener<Integer, String>() {
@Override
public void onMessage(ConsumerRecord<Integer, String> message) {
// The 1st message per partition fails
if (counts.get(message.partition()).incrementAndGet() < 2) {
throw new RuntimeException("Failure wile processing message");
}
}
});
containerProps.setSyncCommits(true);
containerProps.setAckMode(AckMode.RECORD);
final CountDownLatch rebalanceLatch = new CountDownLatch(2);
containerProps.setConsumerRebalanceListener(new ConsumerRebalanceListener() {
@Override
public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
}
@Override
public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
logger.info("manual ack: assigned " + partitions);
rebalanceLatch.countDown();
}
});
CountDownLatch stubbingComplete1 = new CountDownLatch(1);
KafkaMessageListenerContainer<Integer, String> container1 =
spyOnContainer(new KafkaMessageListenerContainer<>(cf, containerProps), stubbingComplete1);
container1.setBeanName("testRebalanceAfterFailedRecord");
container1.start();
Consumer<?, ?> containerConsumer = spyOnConsumer(container1);
final CountDownLatch commitLatch = new CountDownLatch(2);
willAnswer(invocation -> {
@SuppressWarnings({ "unchecked" })
Map<TopicPartition, OffsetAndMetadata> map = invocation.getArgument(0);
try {
return invocation.callRealMethod();
}
finally {
for (Entry<TopicPartition, OffsetAndMetadata> entry : map.entrySet()) {
// Decrement when the last (successful) has been committed
if (entry.getValue().offset() == 2) {
commitLatch.countDown();
}
}
}
}).given(containerConsumer).commitSync(any());
stubbingComplete1.countDown();
ContainerTestUtils.waitForAssignment(container1, embeddedKafka.getPartitionsPerTopic());
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic18);
template.sendDefault(0, 0, "foo");
template.sendDefault(1, 0, "baz");
template.sendDefault(0, 0, "bar");
template.sendDefault(1, 0, "qux");
template.flush();
// Wait until both partitions have committed offset 2 (i.e. the last message)
assertThat(commitLatch.await(30, TimeUnit.SECONDS)).isTrue();
// Start a 2nd consumer, triggering a rebalance
KafkaMessageListenerContainer<Integer, String> container2 =
new KafkaMessageListenerContainer<>(cf, containerProps);
container2.setBeanName("testRebalanceAfterFailedRecord2");
container2.start();
// Wait until both consumers have finished rebalancing
assertThat(rebalanceLatch.await(60, TimeUnit.SECONDS)).isTrue();
// Stop both consumers
container1.stop();
container2.stop();
Consumer<Integer, String> consumer = cf.createConsumer();
consumer.assign(Arrays.asList(new TopicPartition(topic18, 0), new TopicPartition(topic18, 1)));
// Verify that offset of both partitions is the highest committed offset
assertThat(consumer.position(new TopicPartition(topic18, 0))).isEqualTo(2);
assertThat(consumer.position(new TopicPartition(topic18, 1))).isEqualTo(2);
consumer.close();
logger.info("Stop rebalance after failed record");
}
private Consumer<?, ?> spyOnConsumer(KafkaMessageListenerContainer<Integer, String> container) {
Consumer<?, ?> consumer = spy(
KafkaTestUtils.getPropertyValue(container, "listenerConsumer.consumer", Consumer.class));
@@ -1526,8 +1626,10 @@ public class KafkaMessageListenerContainerTests {
return consumer;
}
private KafkaMessageListenerContainer<Integer, String> spyOnContainer(KafkaMessageListenerContainer<Integer, String> container,
private KafkaMessageListenerContainer<Integer, String> spyOnContainer(
KafkaMessageListenerContainer<Integer, String> container,
final CountDownLatch stubbingComplete) {
KafkaMessageListenerContainer<Integer, String> spy = spy(container);
willAnswer(i -> {
if (stubbingComplete.getCount() > 0 && Thread.currentThread().getName().endsWith("-C-1")) {