davsclaus commented on code in PR #27391:
URL: https://github.com/apache/camel/pull/27391#discussion_r4191866962
##########
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:
Minor: retries never stop, and each attempt logs WARN with the full stack
trace, so a broker that is down for an hour gives about 720 stack traces.
Suggest logging the stack trace once and then WARN without it, or using
`getExceptionHandler()` like camel-master.
##########
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:
A leadership event that arrives while a start is in flight is dropped
(`starting` is true, so `startConsumer` returns early). When the in-flight
start then gets here because the generation changed, it stops its consumer and
schedules nothing. If the node is the leader again by then, it holds leadership
with no consumer, which is the state this PR sets out to fix. This happens when:
- a retry is running on the `ZooKeeperMasterRetry` thread and ZooKeeper
suspends and reconnects; or
- the route is stopped and started while the old group thread is still in a
slow `createConsumer` or `start`.
Suggestion: after stopping the consumer here, if `isRunAllowed() &&
isLeader() && delegate == null`, schedule `startConsumer` for the current
generation with no delay. Alternatively, remember the skipped event in a
`pendingStart` flag. A test with a disconnect and reconnect while a retry is
held would pin this down.
##########
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:
Minor: the hard-coded delay makes `testFailedStartIsRetried` take at least 5
s. A package-private setter would let the test use a short delay.
--
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]
