From f2ab4a07c65a19ac23326b1d1e734cf6a9a6a422 Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Wed, 28 Aug 2019 13:55:58 -0400 Subject: [PATCH] GH-70: Support Batch Listeners Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/70 --- docs/src/main/asciidoc/overview.adoc | 11 ++++ .../kafka/KafkaMessageChannelBinder.java | 8 ++- .../stream/binder/kafka/KafkaBinderTests.java | 56 +++++++++++++++++++ 3 files changed, 72 insertions(+), 3 deletions(-) diff --git a/docs/src/main/asciidoc/overview.adoc b/docs/src/main/asciidoc/overview.adoc index c6726d3f6..311e82dde 100644 --- a/docs/src/main/asciidoc/overview.adoc +++ b/docs/src/main/asciidoc/overview.adoc @@ -266,6 +266,17 @@ Ignored if `replicas-assignments` is present. + Default: none (the binder-wide default of 1 is used). +==== Consuming Batches + +Starting with version 3.0, when `spring.cloud.stream.binding..consumer.batch-mode` is set to `true`, all of the records received by polling the Kafka `Consumer` will be presented as a `List` to the listener method. +Otherwise, the method will be called with one record at a time. +The size of the batch is controlled by Kafka consumer properties `max.poll.records`, `min.fetch.bytes`, `fetch.max.wait.ms`; refer to the Kafka documentation for more information. + +IMPORTANT: Retry within the binder is not supported when using batch mode, so `maxAttempts` will be overridden to 1. +You can configure a `SeekToCurrentBatchErrorHandler` (using a `ListenerContainerCustomizer`) to achieve similar functionality to retry in the binder. +You can also use a manual `AckMode` and call `Ackowledgment.nack(index, sleep)` to commit the offsets for a partial batch and have the remaining records redelivered. +Refer to the https://docs.spring.io/spring-kafka/docs/2.3.0.BUILD-SNAPSHOT/reference/html/#committing-offsets[Spring for Apache Kafka documentation] for more information about these techniques. + [[kafka-producer-properties]] ==== Kafka Producer Properties diff --git a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java index 02f4c500d..fbf2f5f0d 100644 --- a/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java +++ b/spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java @@ -83,6 +83,7 @@ import org.springframework.integration.acks.AcknowledgmentCallback; import org.springframework.integration.channel.AbstractMessageChannel; import org.springframework.integration.core.MessageProducer; import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter; +import org.springframework.integration.kafka.inbound.KafkaMessageDrivenChannelAdapter.ListenerMode; import org.springframework.integration.kafka.inbound.KafkaMessageSource; import org.springframework.integration.kafka.outbound.KafkaProducerMessageHandler; import org.springframework.integration.kafka.support.RawRecordHeaderErrorMessageStrategy; @@ -605,15 +606,16 @@ public class KafkaMessageChannelBinder extends this.getContainerCustomizer().configure(messageListenerContainer, destination.getName(), group); // @checkstyle:off - final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = new KafkaMessageDrivenChannelAdapter<>( - messageListenerContainer); + final KafkaMessageDrivenChannelAdapter kafkaMessageDrivenChannelAdapter = + new KafkaMessageDrivenChannelAdapter<>(messageListenerContainer, + extendedConsumerProperties.isBatchMode() ? ListenerMode.batch : ListenerMode.record); // @checkstyle:on kafkaMessageDrivenChannelAdapter .setMessageConverter(getMessageConverter(extendedConsumerProperties)); kafkaMessageDrivenChannelAdapter.setBeanFactory(this.getBeanFactory()); ErrorInfrastructure errorInfrastructure = registerErrorInfrastructure(destination, consumerGroup, extendedConsumerProperties); - if (extendedConsumerProperties.getMaxAttempts() > 1) { + if (!extendedConsumerProperties.isBatchMode() && extendedConsumerProperties.getMaxAttempts() > 1) { kafkaMessageDrivenChannelAdapter .setRetryTemplate(buildRetryTemplate(extendedConsumerProperties)); kafkaMessageDrivenChannelAdapter diff --git a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java index d7d3d5b56..ff4a40b11 100644 --- a/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java +++ b/spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java @@ -480,6 +480,62 @@ public class KafkaBinderTests extends consumerBinding.unbind(); } + @Test + @SuppressWarnings({ "unchecked", "rawtypes" }) + public void testSendAndReceiveBatch() throws Exception { + Binder binder = getBinder(); + BindingProperties outputBindingProperties = createProducerBindingProperties( + createProducerProperties()); + DirectChannel moduleOutputChannel = createBindableChannel("output", + outputBindingProperties); + ExtendedConsumerProperties consumerProperties = createConsumerProperties(); + consumerProperties.setBatchMode(true); + consumerProperties.getExtension().getConfiguration().put("fetch.min.bytes", "1000"); + consumerProperties.getExtension().getConfiguration().put("fetch.max.wait.ms", "5000"); + consumerProperties.getExtension().getConfiguration().put("max.poll.records", "2"); + DirectChannel moduleInputChannel = createBindableChannel("input", + createConsumerBindingProperties(consumerProperties)); + + Binding producerBinding = binder.bindProducer("c.batching", + moduleOutputChannel, outputBindingProperties.getProducer()); + Binding consumerBinding = binder.bindConsumer("c.batching", + "testSendAndReceiveBatch", moduleInputChannel, consumerProperties); + Message message = org.springframework.integration.support.MessageBuilder + .withPayload("foo".getBytes(StandardCharsets.UTF_8)) + .setHeader(KafkaHeaders.PARTITION_ID, 0) + .build(); + + // Let the consumer actually bind to the producer before sending a msg + binderBindUnbindLatency(); + moduleOutputChannel.send(message); + message = MessageBuilder + .withPayload("bar".getBytes(StandardCharsets.UTF_8)) + .setHeader(KafkaHeaders.PARTITION_ID, 0) + .build(); + moduleOutputChannel.send(message); + CountDownLatch latch = new CountDownLatch(1); + AtomicReference>> inboundMessageRef = new AtomicReference<>(); + moduleInputChannel.subscribe(message1 -> { + try { + inboundMessageRef.compareAndSet(null, (Message>) message1); + } + finally { + latch.countDown(); + } + }); + Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message"); + + assertThat(inboundMessageRef.get()).isNotNull(); + List payload = inboundMessageRef.get().getPayload(); + assertThat(payload.get(0)).isEqualTo("foo".getBytes()); + if (payload.size() > 1) { // it's a race as to whether we'll get them both or just one. + assertThat(payload.get(1)).isEqualTo("bar".getBytes()); + } + + producerBinding.unbind(); + consumerBinding.unbind(); + } + @Test @SuppressWarnings("unchecked") public void testDlqWithNativeSerializationEnabledOnDlqProducer() throws Exception {