allthingssecurity commented on code in PR #27391:
URL: https://github.com/apache/camel/pull/27391#discussion_r4191976364
##########
components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java:
##########
@@ -144,40 +186,102 @@ protected void doSuspend() throws Exception {
}
protected Runnable onLockOwned() {
- return () -> {
- if (delegate == null) {
- try {
- // ensure endpoint is also started
- LOG.info("Elected as master. Starting consumer: {}",
endpoint.getConsumerEndpoint());
- ServiceHelper.startService(endpoint.getConsumerEndpoint());
-
- delegate =
endpoint.getConsumerEndpoint().createConsumer(processor);
- delegateService = null;
- if (delegate instanceof SuspendableService) {
- delegateService = (SuspendableService) delegate;
- }
-
- // Lets show we are starting the consumer.
- thisNodeState = createNodeState();
- thisNodeState.setStarted(true);
- groupListener.updateState(thisNodeState);
-
- ServiceHelper.startService(delegate);
- } catch (Exception e) {
- LOG.error("Failed to start master consumer for: {}",
endpoint, e);
- }
+ return () -> startConsumer(-1);
+ }
+ /**
+ * Starts the delegate consumer, as this node holds the leadership.
+ *
+ * @param retryGeneration -1 for a leadership event, else the generation
of the start that failed
+ */
+ private void startConsumer(long retryGeneration) {
+ long startGeneration;
+ CamelNodeState state = null;
+ leadershipLock.lock();
+ try {
+ if (retryGeneration >= 0) {
+ retryTask = null;
+ }
+ if (delegate != null || starting || retryTask != null ||
!isRunAllowed()
+ || retryGeneration >= 0 && retryGeneration != generation
|| !isLeader()) {
+ return;
+ }
+ starting = true;
+ startGeneration = generation;
+ if (startedState == null) {
+ startedState = createNodeState();
+ startedState.setStarted(true);
+ state = startedState;
+ }
+ } finally {
+ leadershipLock.unlock();
+ }
+
+ // the consumer is created and started without holding the lock, as a
start can take long (it may connect to
+ // a broker or a server), and the lock is needed to handle a
disconnect or a stop meanwhile
+ Consumer consumer = null;
+ Exception cause = null;
+ try {
+ // ensure endpoint is also started
+ LOG.info("Elected as master. Starting consumer: {}",
endpoint.getConsumerEndpoint());
+ ServiceHelper.startService(endpoint.getConsumerEndpoint());
+
+ consumer =
endpoint.getConsumerEndpoint().createConsumer(processor);
+
+ // Lets show we are starting the consumer.
+ if (state != null) {
+ thisNodeState = state;
+ groupListener.updateState(state);
+ }
+
+ ServiceHelper.startService(consumer);
+ } catch (Exception e) {
+ cause = e;
+ }
+
+ leadershipLock.lock();
+ try {
+ starting = false;
+ if (startGeneration != generation || !isRunAllowed()) {
+ // disconnected (no longer the master) or stopping while the
consumer started: no event is coming to
+ // stop it later
+ LOG.info("Lost the leadership or stopping while the consumer
started. Stopping consumer: {}",
Review Comment:
Thanks for the careful review. Confirmed and fixed in a70266ff9c98. When the
in-flight start ends after the generation changed, it now schedules a start for
the current generation without delay if `isRunAllowed() && isLeader() &&
delegate == null && retryTask == null` (your first suggestion). It goes through
the retry executor, so a disconnect or stop still cancels it, and an event that
arrives meanwhile sees `retryTask` and skips.
New tests, both fail with the reschedule disabled (`expected: <1> but was:
<0> within 20 seconds`) and pass with it:
- `testReconnectedWhileRetryIsStarting`: the first start fails, the retry is
held in `createConsumer`, then DISCONNECTED and a re-elected CHANGED.
- `testConsumerRestartedWhileCreatingConsumer`: stop and start of the same
consumer instance while the old group thread is in a slow create.
One correction on the second case: `stopRoute`/`startRoute` creates a new
`MasterConsumer` (the route gathers its services again after `setUpDone` is
reset), whose `starting` is false, so a route restart is not affected. It needs
a stop/start of the same instance, e.g. through the JMX consumer operations.
I also extended the TLA+ model (separate group and retry threads,
re-election after a disconnect, consumer restart): the first commit violates "a
leader that runs has a consumer" and its liveness form ("a leader with
runAllowed eventually has a consumer") in both cases; the follow-up satisfies
both (up to 2 failed starts plus a reconnect and a restart). Module suite: 16
tests pass.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java:
##########
@@ -144,40 +186,102 @@ protected void doSuspend() throws Exception {
}
protected Runnable onLockOwned() {
- return () -> {
- if (delegate == null) {
- try {
- // ensure endpoint is also started
- LOG.info("Elected as master. Starting consumer: {}",
endpoint.getConsumerEndpoint());
- ServiceHelper.startService(endpoint.getConsumerEndpoint());
-
- delegate =
endpoint.getConsumerEndpoint().createConsumer(processor);
- delegateService = null;
- if (delegate instanceof SuspendableService) {
- delegateService = (SuspendableService) delegate;
- }
-
- // Lets show we are starting the consumer.
- thisNodeState = createNodeState();
- thisNodeState.setStarted(true);
- groupListener.updateState(thisNodeState);
-
- ServiceHelper.startService(delegate);
- } catch (Exception e) {
- LOG.error("Failed to start master consumer for: {}",
endpoint, e);
- }
+ return () -> startConsumer(-1);
+ }
+ /**
+ * Starts the delegate consumer, as this node holds the leadership.
+ *
+ * @param retryGeneration -1 for a leadership event, else the generation
of the start that failed
+ */
+ private void startConsumer(long retryGeneration) {
+ long startGeneration;
+ CamelNodeState state = null;
+ leadershipLock.lock();
+ try {
+ if (retryGeneration >= 0) {
+ retryTask = null;
+ }
+ if (delegate != null || starting || retryTask != null ||
!isRunAllowed()
+ || retryGeneration >= 0 && retryGeneration != generation
|| !isLeader()) {
+ return;
+ }
+ starting = true;
+ startGeneration = generation;
+ if (startedState == null) {
+ startedState = createNodeState();
+ startedState.setStarted(true);
+ state = startedState;
+ }
+ } finally {
+ leadershipLock.unlock();
+ }
+
+ // the consumer is created and started without holding the lock, as a
start can take long (it may connect to
+ // a broker or a server), and the lock is needed to handle a
disconnect or a stop meanwhile
+ Consumer consumer = null;
+ Exception cause = null;
+ try {
+ // ensure endpoint is also started
+ LOG.info("Elected as master. Starting consumer: {}",
endpoint.getConsumerEndpoint());
+ ServiceHelper.startService(endpoint.getConsumerEndpoint());
+
+ consumer =
endpoint.getConsumerEndpoint().createConsumer(processor);
+
+ // Lets show we are starting the consumer.
+ if (state != null) {
+ thisNodeState = state;
+ groupListener.updateState(state);
+ }
+
+ ServiceHelper.startService(consumer);
+ } catch (Exception e) {
+ cause = e;
+ }
+
+ leadershipLock.lock();
+ try {
+ starting = false;
+ if (startGeneration != generation || !isRunAllowed()) {
+ // disconnected (no longer the master) or stopping while the
consumer started: no event is coming to
+ // stop it later
+ LOG.info("Lost the leadership or stopping while the consumer
started. Stopping consumer: {}",
+ endpoint.getConsumerEndpoint());
+ ServiceHelper.stopAndShutdownServices(consumer);
+ } else if (cause != null) {
+ // forget the consumer (its start stopped it) and try again
later, as long as this node is the master
+ LOG.warn("Failed to start master consumer for: {}. Trying
again in {} millis.", endpoint,
Review Comment:
Done in a70266ff9c98: a failed start now goes to
`getExceptionHandler().handleException(...)`, as camel-master does, with the
attempt number in the message (`... (attempt #3). Trying again in 5000
millis.`), and the counter resets on success, disconnect or stop. As in
camel-master, the default `LoggingExceptionHandler` still logs each attempt at
WARN with the stack trace; the endpoint's `exceptionHandler` /
`bridgeErrorHandler` options can now route or quiet them.
_Claude Code on behalf of allthingssecurity_
##########
components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java:
##########
@@ -36,13 +42,28 @@
@ManagedResource(description = "Managed ZooKeeper Master Consumer")
public class MasterConsumer extends DefaultConsumer {
private static final transient Logger LOG =
LoggerFactory.getLogger(MasterConsumer.class);
+ // delay before the master tries again to start a consumer whose start
failed
+ private static final long RETRY_DELAY_MILLIS = 5000;
Review Comment:
Done in a70266ff9c98: package-private `setRetryDelay(long)` (default still 5
s). The retry tests use 10 ms, and `MasterConsumerLeadershipTest` now runs in
about 0.6 s instead of 5 s.
_Claude Code on behalf of allthingssecurity_
--
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]
