From 48b5c2c8b04db642eb031814da5afa4590adc592 Mon Sep 17 00:00:00 2001 From: Axel Uhl Date: Tue, 3 Feb 2026 11:34:37 +0100 Subject: [PATCH] bug6211: use explicit ACKs in RabbitInputStreamProvider and set basicQos(1); this to avoid any flooding of TCP sockets by the RabbitMQ server when the client isn't able to consume traffic fast enough --- .../sap/sse/replication/impl/RabbitInputStreamProvider.java | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/RabbitInputStreamProvider.java b/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/RabbitInputStreamProvider.java index 0f228cd726e..550d44a6343 100755 --- a/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/RabbitInputStreamProvider.java +++ b/java/com.sap.sse.replication/src/com/sap/sse/replication/impl/RabbitInputStreamProvider.java @@ -39,14 +39,16 @@ public class RabbitInputStreamProvider extends NamedImpl { messagesAreWrittenToThis = new PipedOutputStream(); clientReadsFromThis = new PipedInputStream(messagesAreWrittenToThis); final QueueingConsumer messageConsumer = new QueueingConsumer(channel); - channel.basicConsume(queueName, /* auto-ack */ true, messageConsumer); + channel.basicQos(1); // with "manual" acknowledgement, process one message at a time, avoiding overloading inbound socket + channel.basicConsume(queueName, /* auto-ack */ false, messageConsumer); new Thread(getClass().getSimpleName()) { @Override public void run() { while (true) { try { - Delivery delivery = messageConsumer.nextDelivery(); + final Delivery delivery = messageConsumer.nextDelivery(); byte[] bytesFromMessage = delivery.getBody(); + channel.basicAck(delivery.getEnvelope().getDeliveryTag(), /* multiple */ false); if (RabbitOutputStream.startsWithTerminationCommand(bytesFromMessage, bytesFromMessage.length)) { if (bytesFromMessage.length == RabbitOutputStream.TERMINATION_COMMAND.length) { // received exactly TERMINATION_COMMAND