From 01a715a9a960b23b9f14d35b301b62ab1a935a9f Mon Sep 17 00:00:00 2001 From: Marius Bogoevici Date: Wed, 4 Jan 2017 22:37:47 -0500 Subject: [PATCH] Observe `bufferSize` setting Fixes #77 Signed-off-by: Marius Bogoevici --- .../cloud/stream/binder/kafka/KafkaMessageChannelBinder.java | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) 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 62947b3d4..241ee8501 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 @@ -247,8 +247,7 @@ public class KafkaMessageChannelBinder extends } props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, this.configurationProperties.getKafkaConnectionString()); props.put(ProducerConfig.RETRIES_CONFIG, 0); - props.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384); - props.put(ProducerConfig.LINGER_MS_CONFIG, 1); + props.put(ProducerConfig.BATCH_SIZE_CONFIG, String.valueOf(producerProperties.getExtension().getBufferSize())); props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432); props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class); props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, ByteArraySerializer.class);