[durdin85]

Description

This PR makes the RabbitMQ event bus keep events that are published while the
AMQP broker is unreachable and send them, in the order they were published, once
the connection is back. Until now such events were discarded.

Behaviour:

• Events that cannot be sent are kept in an in-memory FIFO buffer (routing key,
  body, time of buffering).
• Buffering starts only after the broker was reachable at least once. Before
  that, publish() fails as it does today.
• The buffer is bounded: at most bufferSize events (default 1000, the oldest is
  dropped when full) and each event is dropped bufferExpiry milliseconds
  (default 300000) after it was buffered. bufferSize 0 disables buffering and
  restores the previous behaviour.
• While the buffer is not empty, or the connection is down, new events are
  appended to the buffer, so they never overtake older ones. publish() returns
  normally for a buffered event and no longer connects on the calling thread
  while the broker is down (the connection timeout of amqp-client 5.17.0 is
  60 s).
• ReconnectionTask now runs for the lifetime of the event bus instead of being
  started by DisconnectHandler. Every retryInterval it drops expired events,
  reconnects if there is no connection and sends the buffered events. This also
  re-establishes connections closed by the bus itself (AlreadyClosedException
  in publish(), blocked connection), which previously started no reconnect.
• stop() shuts the executor down so the task ends with the bus.

Both limits are optional properties of the existing bean, next to the other AMQP
settings:

<bean id="eventNotificationBus" class="org.apache.cloudstack.mom.rabbitmq.RabbitMQEventBus">
    ...
    <property name="bufferSize" value="1000"/>
    <property name="bufferExpiry" value="300000"/>
</bean>

Known limits, by design:

• No publisher confirms are used. An event written to the socket just before the
  connection breaks can still be lost unnoticed.
• An event whose send fails after the broker already received it is sent again
  after the reconnect (at-least-once for buffered events).
• The buffer is in memory only and is lost when the management server restarts.

No new dependencies; the change is confined to RabbitMQEventBus.

Fixes: #14676

AI disclosure: The code, the unit test and this description were generated with an AI assistant. The design and its constraints were set by the submitter and the generated code had been reviewed by the submitter. If the project does not accept AI-generated contributions, please say so and treat this as a proposal for discussion.

Types of changes

• Breaking change (fix or feature that would cause existing functionality to change)
• New feature (non-breaking change which adds functionality)
• Bug fix (non-breaking change which fixes an issue)
• Enhancement (improves an existing feature and functionality)
• Cleanup (Code refactoring and cleanup, that may add test cases)
• Build/CI
• Test (unit or integration test code)

Feature/Enhancement Scale or Bug Severity

Feature/Enhancement Scale

• Major
• Minor

How Has This Been Tested?

New unit test RabbitMQEventBusTest (mocked AMQP connection and channel):

• publish() still fails when the broker was never reachable;
• events published while disconnected are sent in order after the reconnect, and
  an event published while the buffer is not empty does not overtake them;
• an event whose send fails with an IOException is buffered and re-sent;
• the oldest event is dropped when the buffer is full;
• expired events are dropped, also while disconnected.

The plugin builds and RabbitMQEventBusTest passes on main and on the 4.22 branch (mvn -pl plugins/event-bus/rabbitmq -am test -Dtest=RabbitMQEventBusTest).

Not yet tested against a real RabbitMQ broker.

How did you try to break this feature and the system with this change?

The unit tests cover a full buffer, expired events, a failing send and a
connection that returns while events are still buffered.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache CloudStack community! If you have any issues or are unsure about any anything please check our Contribution Guide (https://github.com/apache/cloudstack/blob/main/CONTRIBUTING.md)
Here are some useful points:

• In case of a new feature add useful documentation (raise doc PR at https://github.com/apache/cloudstack-documentation)
• Be patient and persistent. It might take some time to get a review or get the final approval from the committers.
• Pay attention to the quality of your code, ensure tests are passing and your PR doesn't have conflicts.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Issues, Mailing list and Slack.
• Be sure to read the CloudStack Coding Conventions.
  Apache CloudStack is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@cloudstack.apache.org (https://cloudstack.apache.org/mailing-lists.html)
  Slack: https://apachecloudstack.slack.com/