Skip to content

Rabbitmq eventbus buffer implementation #14677

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
durdin85 wants to merge 3 commits into
base: main
Choose a base branch
from
Open

Rabbitmq eventbus buffer implementation #14677

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
e55b4a4
rabbitmq event bus: buffer events while the broker is unreachable
durdin85 Nov 11, 2026
3220076
rabbitmq event bus: harden the reconnect loop
durdin85 Nov 11, 2026
0b85b3d
rabbitmq event bus: log when events are buffered and re-sent
durdin85 Nov 11, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
146 changes: 129 additions & 17 deletions ...event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import java.net.ConnectException;
import java.security.KeyManagementException;
import java.security.NoSuchAlgorithmException;
import java.util.ArrayDeque;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
Expand Down Expand Up @@ -85,11 +86,23 @@ public static void setUseSsl(String useSsl) {

private static Integer retryInterval;

// max number of events kept in memory while the AMQP server is unreachable, 0 disables buffering
private static Integer bufferSize = 1000;

// time in milliseconds after which a buffered event is dropped if it could not be sent
private static Integer bufferExpiry = 300000;

// events (routing key, description, time of buffering) waiting to be sent, oldest first
private static final ArrayDeque<Ternary<String, String, Long>> s_buffer = new ArrayDeque<>();

// events are buffered only after the AMQP server was reachable at least once
private static volatile boolean s_connectedOnce = false;

// hashmap to book keep the registered subscribers
private static ConcurrentHashMap<String, Ternary<String, Channel, EventSubscriber>> s_subscribers;

// connection to AMQP server,
private static Connection s_connection = null;
private static volatile Connection s_connection = null;

// AMQP server should consider messages acknowledged once delivered if _autoAck is true
private static boolean s_autoAck = true;
Expand Down Expand Up @@ -177,6 +190,14 @@ public static void setRetryInterval(Integer retryInterval) {
RabbitMQEventBus.retryInterval = retryInterval;
}

public static void setBufferSize(Integer bufferSize) {
RabbitMQEventBus.bufferSize = bufferSize;
}

public static void setBufferExpiry(Integer bufferExpiry) {
RabbitMQEventBus.bufferExpiry = bufferExpiry;
}

/** Call to subscribe to interested set of events
*
* @param topic defines category and type of the events being subscribed to
Expand Down Expand Up @@ -273,20 +294,106 @@ public void publish(Event event) throws EventBusException {
String routingKey = createRoutingKey(event);
String eventDescription = event.getDescription();

// keep the order: nothing is sent directly while older events are waiting or the connection is down
if (bufferEvent(routingKey, eventDescription, false)) {
return;
}

try {
Connection connection = getConnection();
Channel channel = createChannel(connection);
createExchange(channel, amqpExchangeName);
publishEventToExchange(channel, amqpExchangeName, routingKey, eventDescription);
channel.close();
sendEvent(getConnection(), routingKey, eventDescription);
} catch (AlreadyClosedException e) {
closeConnection();
if (bufferEvent(routingKey, eventDescription, true)) {
return;
}
throw new EventBusException("Failed to publish event to message broker as connection to AMQP broker in lost");
} catch (IOException e) {
if (bufferEvent(routingKey, eventDescription, true)) {
return;
}
throw new EventBusException("Failed to publish event to message broker due to " + e.getMessage());
} catch (Exception e) {
throw new EventBusException("Failed to publish event to message broker due to " + e.getMessage());
}
}

private void sendEvent(Connection connection, String routingKey, String eventDescription) throws Exception {
Channel channel = createChannel(connection);
createExchange(channel, amqpExchangeName);
publishEventToExchange(channel, amqpExchangeName, routingKey, eventDescription);
channel.close();
}

/** adds the event to the buffer, dropping the oldest one if the buffer is full
*
* @param sendFailed false if the event was not tried yet and has to be buffered only to keep the order
* @return true if the event was buffered, false if the caller has to deal with it
*/
private boolean bufferEvent(String routingKey, String eventDescription, boolean sendFailed) {
if (!s_connectedOnce || bufferSize <= 0) {
return false;
}
synchronized (s_buffer) {
if (!sendFailed && s_buffer.isEmpty() && s_connection != null) {
return false;
}
if (s_buffer.isEmpty()) {
logger.warn("Unable to send events to AMQP server, buffering them until they can be sent");
}
if (s_buffer.size() >= bufferSize) {
logger.warn("Event buffer is full, dropping the oldest event " + s_buffer.poll().first());
}
s_buffer.add(new Ternary<>(routingKey, eventDescription, System.currentTimeMillis()));
}
return true;
}

// drops the expired events and, if connected, sends the buffered ones in the order they were added
private void flushBuffer() {
int sent = 0;
while (true) {
Ternary<String, String, Long> bufferedEvent;
synchronized (s_buffer) {
bufferedEvent = s_buffer.peek();
}
if (bufferedEvent == null) {
break;
}

boolean expired = System.currentTimeMillis() - bufferedEvent.third() > bufferExpiry;
Connection connection = s_connection;
if (!expired && connection == null) {
break;
}

if (expired) {
logger.warn("Dropping expired event " + bufferedEvent.first() + " as it could not be sent to AMQP server in time");
} else {
try {
sendEvent(connection, bufferedEvent.first(), bufferedEvent.second());
sent++;
} catch (AlreadyClosedException e) {
closeConnection();
break;
} catch (IOException e) {
break; // retry with the next run
} catch (Exception e) {
logger.warn("Dropping buffered event " + bufferedEvent.first() + " due to " + e.getMessage());
}
}

// the event stays in the buffer while being sent, so that newer events queue up behind it
synchronized (s_buffer) {
if (s_buffer.peek() == bufferedEvent) {
s_buffer.poll();
}
}
}
if (sent > 0) {
logger.info("Sent " + sent + " buffered events to AMQP server");
}
}

/** creates a routing key from the event details.
* created routing key will be used while publishing the message to exchange on AMQP server
*/
Expand Down Expand Up @@ -392,6 +499,7 @@ private synchronized Connection createConnection() throws KeyManagementException
connection.addShutdownListener(disconnectHandler);
connection.addBlockedListener(blockedConnectionHandler);
s_connection = connection;
s_connectedOnce = true;
return s_connection;
}

Expand Down Expand Up @@ -508,6 +616,7 @@ public synchronized boolean stop() {
}

closeConnection();
executorService.shutdownNow();
return true;
}

Expand Down Expand Up @@ -540,35 +649,38 @@ public void shutdownCompleted(ShutdownSignalException shutdownSignalException) {
}

abortConnection(); // disconnected to AMQP server, so abort the connection and channels
// the ReconnectionTask started with the event bus takes care of the re-connect
logger.warn("Connection has been shutdown by AMQP server. Attempting to reconnect.");

// initiate re-connect process
ReconnectionTask reconnect = new ReconnectionTask();
executorService.submit(reconnect);
}
}
}

// retry logic to connect back to AMQP server after loss of connection
// runs for the lifetime of the event bus: connects back to AMQP server after any loss of connection
// and sends the events buffered in the meantime
private class ReconnectionTask extends ManagedContextRunnable {

boolean connected = false;
Connection connection = null;

@Override
protected void runInContext() {

while (!connected) {
while (true) {
try {
Thread.sleep(retryInterval);
} catch (InterruptedException ie) {
// ignore timer interrupts
return; // event bus is being stopped
}

try {
flushBuffer();
try {
connection = createConnection();
connected = true;
// check and connect under the same lock as getConnection(), so a connection is never replaced
synchronized (RabbitMQEventBus.this) {
if (s_connection != null) {
continue;
}
connection = createConnection();
}
} catch (IOException ie) {
continue; // can't establish connection to AMQP server yet, so continue
}
Expand Down Expand Up @@ -617,11 +729,11 @@ public void handleDelivery(String queueName, Envelope envelope, AMQP.BasicProper
subscriberDetails.second(channel);
s_subscribers.put(subscriberId, subscriberDetails);
}
flushBuffer();
} catch (Exception e) {
logger.warn("Failed to recreate queues and binding for the subscribers due to " + e.getMessage());
}
}
return;
}
}
}
Loading