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

GH-62: Add MANUAL_IMMEDIATE Test with Existing

Browse Source
This commit is contained in:
Gary Russell
2016-05-06 11:20:36 -04:00
parent d032795180
commit 3c6a673950
1 changed files with 47 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -72,9 +72,11 @@ public class ConcurrentMessageListenerContainerTests {
private static String topic6 = "testTopic6";
private static String topic7 = "testTopic7";
@ClassRule
public static KafkaEmbedded embeddedKafka = new KafkaEmbedded(1, true, topic1, topic2, topic3, topic4, topic5,
topic6);
topic6, topic7);
@Test
public void testAutoCommit() throws Exception {
@@ -278,7 +280,7 @@ public class ConcurrentMessageListenerContainerTests {
private void testManualCommitGuts(AckMode ackMode, String topic) throws Exception {
logger.info("Start " + ackMode);
Map<String, Object> props = KafkaTestUtils.consumerProps("test4", "false", embeddedKafka);
Map<String, Object> props = KafkaTestUtils.consumerProps("test" + ackMode, "false", embeddedKafka);
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
ConcurrentMessageListenerContainer<Integer, String> container =
new ConcurrentMessageListenerContainer<>(cf, topic);
@@ -312,6 +314,49 @@ public class ConcurrentMessageListenerContainerTests {
logger.info("Stop " + ackMode);
}
@Test
public void testManualCommitExisting() throws Exception {
logger.info("Start MANUAL_IMMEDIATE with Existing");
Map<String, Object> senderProps = KafkaTestUtils.producerProps(embeddedKafka);
ProducerFactory<Integer, String> pf = new DefaultKafkaProducerFactory<Integer, String>(senderProps);
KafkaTemplate<Integer, String> template = new KafkaTemplate<>(pf);
template.setDefaultTopic(topic7);
template.send(0, "foo");
template.send(2, "bar");
template.send(0, "baz");
template.send(2, "qux");
template.flush();
Map<String, Object> props = KafkaTestUtils.consumerProps("testManualExisting", "false", embeddedKafka);
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
DefaultKafkaConsumerFactory<Integer, String> cf = new DefaultKafkaConsumerFactory<Integer, String>(props);
ConcurrentMessageListenerContainer<Integer, String> container =
new ConcurrentMessageListenerContainer<>(cf, topic7);
final CountDownLatch latch = new CountDownLatch(8);
container.setMessageListener(new AcknowledgingMessageListener<Integer, String>() {
@Override
public void onMessage(ConsumerRecord<Integer, String> message, Acknowledgment ack) {
logger.info("manualExisting: " + message);
ack.acknowledge();
latch.countDown();
}
});
container.setConcurrency(1);
container.setAckMode(AckMode.MANUAL_IMMEDIATE);
container.setBeanName("testManualExisting");
container.start();
ContainerTestUtils.waitForAssignment(container, embeddedKafka.getPartitionsPerTopic());
template.send(0, "fooo");
template.send(2, "barr");
template.send(0, "bazz");
template.send(2, "quxx");
template.flush();
assertThat(latch.await(60, TimeUnit.SECONDS)).isTrue();
container.stop();
logger.info("Stop MANUAL_IMMEDIATE with Existing");
}
@SuppressWarnings("unchecked")
@Test