SpringCloud/spring-cloud-stream
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-70: Support Batch Listeners

Browse Source 
Resolves https://github.com/spring-cloud/spring-cloud-stream-binder-kafka/issues/70
This commit is contained in:
Gary Russell
2019-08-28 13:55:58 -04:00
committed by Soby Chacko
parent 584115580b
commit f2ab4a07c6
3 changed files with 72 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

11
docs/src/main/asciidoc/overview.adoc
View File

@@ -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.<name>.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

8
spring-cloud-stream-binder-kafka/src/main/java/org/springframework/cloud/stream/binder/kafka/KafkaMessageChannelBinder.java
View File

@@ -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

56
spring-cloud-stream-binder-kafka/src/test/java/org/springframework/cloud/stream/binder/kafka/KafkaBinderTests.java
View File

@@ -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<KafkaConsumerProperties> 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<MessageChannel> producerBinding = binder.bindProducer("c.batching",
moduleOutputChannel, outputBindingProperties.getProducer());
Binding<MessageChannel> 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<Message<List<byte[]>>> inboundMessageRef = new AtomicReference<>();
moduleInputChannel.subscribe(message1 -> {
try {
inboundMessageRef.compareAndSet(null, (Message<List<byte[]>>) message1);
}
finally {
latch.countDown();
}
});
Assert.isTrue(latch.await(5, TimeUnit.SECONDS), "Failed to receive message");
assertThat(inboundMessageRef.get()).isNotNull();
List<byte[]> 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 {