From e55b4a40a684a7fe8a7560690d994661a8318ac0 Mon Sep 17 00:00:00 2001 From: durdin85 <6603216+durdin85@users.noreply.github.com> Date: Wed, 11 Nov 2026 11:00:32 +0000 Subject: [PATCH 1/3] rabbitmq event bus: buffer events while the broker is unreachable Events published while the AMQP connection is down were discarded. Keep them in a bounded in-memory FIFO buffer and send them in order once the connection is back. - events are buffered only after the broker was reachable once - buffer is capped (bufferSize, default 1000, oldest dropped first) and entries expire (bufferExpiry, default 300000 ms) - publish() no longer connects on the calling thread while the broker is down - ReconnectionTask runs for the lifetime of the bus, so connections closed by the bus itself are re-established as well This change, including the unit test, was generated with an AI assistant (Claude Code). Fixes: #14676 Co-Authored-By: Claude Fable 5.2 --- .../mom/rabbitmq/RabbitMQEventBus.java | 134 ++++++++++++-- .../mom/rabbitmq/RabbitMQEventBusTest.java | 167 ++++++++++++++++++ 2 files changed, 285 insertions(+), 16 deletions(-) create mode 100644 plugins/event-bus/rabbitmq/src/test/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBusTest.java diff --git a/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java b/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java index b98147b5f031..fd1a7e26fc91 100644 --- a/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java +++ b/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java @@ -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; @@ -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> 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> 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; @@ -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 @@ -273,20 +294,98 @@ 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.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() { + while (true) { + Ternary bufferedEvent; + synchronized (s_buffer) { + bufferedEvent = s_buffer.peek(); + } + if (bufferedEvent == null) { + return; + } + + boolean expired = System.currentTimeMillis() - bufferedEvent.third() > bufferExpiry; + Connection connection = s_connection; + if (!expired && connection == null) { + return; + } + + 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()); + } catch (AlreadyClosedException e) { + closeConnection(); + return; + } catch (IOException e) { + return; // 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(); + } + } + } + } + /** creates a routing key from the event details. * created routing key will be used while publishing the message to exchange on AMQP server */ @@ -392,6 +491,7 @@ private synchronized Connection createConnection() throws KeyManagementException connection.addShutdownListener(disconnectHandler); connection.addBlockedListener(blockedConnectionHandler); s_connection = connection; + s_connectedOnce = true; return s_connection; } @@ -508,6 +608,7 @@ public synchronized boolean stop() { } closeConnection(); + executorService.shutdownNow(); return true; } @@ -540,35 +641,36 @@ 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 + } + + flushBuffer(); + if (s_connection != null) { + continue; } try { try { connection = createConnection(); - connected = true; } catch (IOException ie) { continue; // can't establish connection to AMQP server yet, so continue } @@ -620,8 +722,8 @@ public void handleDelivery(String queueName, Envelope envelope, AMQP.BasicProper } catch (Exception e) { logger.warn("Failed to recreate queues and binding for the subscribers due to " + e.getMessage()); } + flushBuffer(); } - return; } } } diff --git a/plugins/event-bus/rabbitmq/src/test/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBusTest.java b/plugins/event-bus/rabbitmq/src/test/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBusTest.java new file mode 100644 index 000000000000..281e461575d8 --- /dev/null +++ b/plugins/event-bus/rabbitmq/src/test/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBusTest.java @@ -0,0 +1,167 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package org.apache.cloudstack.mom.rabbitmq; + +import java.io.IOException; +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.ArrayDeque; + +import org.apache.cloudstack.framework.events.Event; +import org.apache.cloudstack.framework.events.EventBusException; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentMatchers; +import org.mockito.InOrder; +import org.mockito.Mockito; + +import com.rabbitmq.client.Channel; +import com.rabbitmq.client.Connection; + +public class RabbitMQEventBusTest { + + private RabbitMQEventBus eventBus; + private Connection connection; + private Channel channel; + + @Before + public void setUp() throws Exception { + eventBus = new RabbitMQEventBus(); + RabbitMQEventBus.setExchange("cloudstack-events"); + RabbitMQEventBus.setBufferSize(1000); + RabbitMQEventBus.setBufferExpiry(300000); + + channel = Mockito.mock(Channel.class); + connection = Mockito.mock(Connection.class); + Mockito.when(connection.createChannel()).thenReturn(channel); + + ((ArrayDeque)getField("s_buffer").get(null)).clear(); + setConnection(null); + getField("s_connectedOnce").set(null, true); + } + + @After + public void tearDown() throws Exception { + setConnection(null); + getField("s_connectedOnce").set(null, false); + } + + private Field getField(String name) throws Exception { + Field field = RabbitMQEventBus.class.getDeclaredField(name); + field.setAccessible(true); + return field; + } + + private void setConnection(Connection connection) throws Exception { + getField("s_connection").set(null, connection); + } + + private void flushBuffer() throws Exception { + Method method = RabbitMQEventBus.class.getDeclaredMethod("flushBuffer"); + method.setAccessible(true); + method.invoke(eventBus); + } + + private Event event(String description) { + Event event = new Event("server", "ActionEvent", "VM.START", "VirtualMachine", "uuid"); + event.setDescription(description); + return event; + } + + private void verifyPublished(InOrder inOrder, String description) throws IOException { + inOrder.verify(channel).basicPublish(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), ArgumentMatchers.any(), + ArgumentMatchers.eq(description.getBytes())); + } + + private void verifyPublishedCount(int count) throws IOException { + Mockito.verify(channel, Mockito.times(count)).basicPublish(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), + ArgumentMatchers.any(), ArgumentMatchers.any(byte[].class)); + } + + @Test(expected = EventBusException.class) + public void publishFailsWhenBrokerWasNeverReachable() throws Exception { + getField("s_connectedOnce").set(null, false); + RabbitMQEventBus.setServer("localhost"); + RabbitMQEventBus.setPort(1); + RabbitMQEventBus.setUsername("guest"); + RabbitMQEventBus.setPassword("guest"); + + eventBus.publish(event("1")); + } + + @Test + public void eventsPublishedWhileDisconnectedAreSentInOrderAfterReconnect() throws Exception { + eventBus.publish(event("1")); + eventBus.publish(event("2")); + Mockito.verifyNoInteractions(channel); + + setConnection(connection); + eventBus.publish(event("3")); // must not overtake the buffered ones + Mockito.verifyNoInteractions(channel); + + flushBuffer(); + InOrder inOrder = Mockito.inOrder(channel); + verifyPublished(inOrder, "1"); + verifyPublished(inOrder, "2"); + verifyPublished(inOrder, "3"); + + eventBus.publish(event("4")); // buffer is empty again, sent directly + verifyPublished(inOrder, "4"); + verifyPublishedCount(4); + } + + @Test + public void eventIsBufferedWhenSendingFails() throws Exception { + setConnection(connection); + Mockito.doThrow(new IOException("broken pipe")).doNothing().when(channel) + .basicPublish(ArgumentMatchers.anyString(), ArgumentMatchers.anyString(), ArgumentMatchers.any(), ArgumentMatchers.any(byte[].class)); + + eventBus.publish(event("1")); + verifyPublishedCount(1); + + flushBuffer(); + verifyPublishedCount(2); + } + + @Test + public void oldestEventIsDroppedWhenBufferIsFull() throws Exception { + RabbitMQEventBus.setBufferSize(2); + eventBus.publish(event("1")); + eventBus.publish(event("2")); + eventBus.publish(event("3")); + + setConnection(connection); + flushBuffer(); + InOrder inOrder = Mockito.inOrder(channel); + verifyPublished(inOrder, "2"); + verifyPublished(inOrder, "3"); + verifyPublishedCount(2); + } + + @Test + public void expiredEventsAreDropped() throws Exception { + RabbitMQEventBus.setBufferExpiry(-1); + eventBus.publish(event("1")); + + flushBuffer(); // expires also without a connection + setConnection(connection); + flushBuffer(); + Mockito.verifyNoInteractions(channel); + } +} From 32200761ac1c628c566a3b23f5a3d52faeb81397 Mon Sep 17 00:00:00 2001 From: durdin85 <6603216+durdin85@users.noreply.github.com> Date: Wed, 11 Nov 2026 11:26:53 +0000 Subject: [PATCH 2/3] rabbitmq event bus: harden the reconnect loop - keep the buffer flush inside the try block, so an unexpected exception cannot end the only reconnect thread - check for an existing connection and connect under the same lock as getConnection(), so a connection created by publish() is not replaced and leaked Generated with an AI assistant (Claude Code). Co-Authored-By: Claude Fable 5.2 --- .../mom/rabbitmq/RabbitMQEventBus.java | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java b/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java index fd1a7e26fc91..b468710e7aad 100644 --- a/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java +++ b/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java @@ -663,14 +663,16 @@ protected void runInContext() { return; // event bus is being stopped } - flushBuffer(); - if (s_connection != null) { - continue; - } - try { + flushBuffer(); try { - connection = createConnection(); + // 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 } @@ -719,10 +721,10 @@ 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()); } - flushBuffer(); } } } From 0b85b3ddc3d9f847f1c3a34af3ae9c497e24ba15 Mon Sep 17 00:00:00 2001 From: durdin85 <6603216+durdin85@users.noreply.github.com> Date: Wed, 11 Nov 2026 12:52:11 +0000 Subject: [PATCH 3/3] rabbitmq event bus: log when events are buffered and re-sent Generated with an AI assistant (Claude Code). Co-Authored-By: Claude Fable 5.2 --- .../mom/rabbitmq/RabbitMQEventBus.java | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java b/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java index b468710e7aad..d7742ee32266 100644 --- a/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java +++ b/plugins/event-bus/rabbitmq/src/main/java/org/apache/cloudstack/mom/rabbitmq/RabbitMQEventBus.java @@ -337,6 +337,9 @@ private boolean bufferEvent(String routingKey, String eventDescription, boolean 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()); } @@ -347,19 +350,20 @@ private boolean bufferEvent(String routingKey, String eventDescription, boolean // 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 bufferedEvent; synchronized (s_buffer) { bufferedEvent = s_buffer.peek(); } if (bufferedEvent == null) { - return; + break; } boolean expired = System.currentTimeMillis() - bufferedEvent.third() > bufferExpiry; Connection connection = s_connection; if (!expired && connection == null) { - return; + break; } if (expired) { @@ -367,11 +371,12 @@ private void flushBuffer() { } else { try { sendEvent(connection, bufferedEvent.first(), bufferedEvent.second()); + sent++; } catch (AlreadyClosedException e) { closeConnection(); - return; + break; } catch (IOException e) { - return; // retry with the next run + break; // retry with the next run } catch (Exception e) { logger.warn("Dropping buffered event " + bufferedEvent.first() + " due to " + e.getMessage()); } @@ -384,6 +389,9 @@ private void flushBuffer() { } } } + if (sent > 0) { + logger.info("Sent " + sent + " buffered events to AMQP server"); + } } /** creates a routing key from the event details.