durdin85 opened a new pull request, #14677:
URL: https://github.com/apache/cloudstack/pull/14677
### 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:
```xml
<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)
- [x] 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
- [x] 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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
