From 0cad02e8169f46889494bb7415c0a008badf2b49 Mon Sep 17 00:00:00 2001 From: addozhang Date: Wed, 20 Sep 2017 17:53:21 +0800 Subject: [PATCH] GH-425: testManualCommitEx: set syncCommit=false Resolves spring-projects/spring-kafka#425 To really perform `commitCallback` via `commitAsync()` the `commitSync` must be set to `false` for the `ContainerProperties` **Cherry-pick to 1.3 and master** --- .../listener/ConcurrentMessageListenerContainerTests.java | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java index 8514d3aa..e3021f20 100644 --- a/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java +++ b/spring-kafka/src/test/java/org/springframework/kafka/listener/ConcurrentMessageListenerContainerTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2016 the original author or authors. + * Copyright 2016-2017 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -44,7 +44,6 @@ import org.apache.kafka.clients.consumer.ConsumerRebalanceListener; import org.apache.kafka.clients.consumer.ConsumerRecords; 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; @@ -291,7 +290,6 @@ public class ConcurrentMessageListenerContainerTests { } @Test - @Ignore // TODO https://github.com/spring-projects/spring-kafka/issues/62 using SYNC for avoidance public void testManualCommitExisting() throws Exception { this.logger.info("Start MANUAL_IMMEDIATE with Existing"); Map senderProps = KafkaTestUtils.producerProps(embeddedKafka); @@ -314,7 +312,7 @@ public class ConcurrentMessageListenerContainerTests { latch.countDown(); }); containerProps.setAckMode(AckMode.MANUAL_IMMEDIATE); - + containerProps.setSyncCommits(false); final CountDownLatch commits = new CountDownLatch(8); final AtomicReference exceptionRef = new AtomicReference<>(); containerProps.setCommitCallback((offsets, exception) -> {