You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by "allthingssecurity (via GitHub)" <gi...@apache.org> on 2026/10/06 00:04:25 UTC
[PR] CAMEL-25315: camel-zookeeper-master - retry a failed start of the master consumer, and do not start it after the leadership was lost or the route stopped [camel]
allthingssecurity opened a new pull request, #27391:
URL: https://github.com/apache/camel/pull/27391
# Description
[CAMEL-25315](https://issues.apache.org/jira/browse/CAMEL-25315)
`MasterConsumer` (camel-zookeeper-master) starts the delegate consumer from the `CHANGED` group event (group thread) and stops it from `DISCONNECTED` (Curator connection-state thread, and the closing thread in `ZooKeeperGroup.close()`) and from `doStop`, with no synchronisation and only `if (delegate == null)`:
- a delegate whose start failed (broker or server not reachable when the node is elected) stays in the field, so no later event starts it again: the node keeps the leadership and nobody consumes until it loses its connection or the route is restarted;
- a disconnect or a stop between the null check and the assignment of the new delegate stops nothing, and the delegate is then created and started: it keeps consuming on a node that is no longer the master (also after reconnecting as a standby), or after the route stopped (when the start takes longer than the 5 s `close()` waits for the group thread).
This change adds a `leadershipLock` and a generation counter incremented by `onDisconnected` and `doStop`. A start is marked under the lock after checking again that the group is connected and this node is the master; the consumer is created and started without holding the lock; it is published under the lock only if the generation did not change, otherwise stopped. A failed start is retried after 5 seconds by a scheduled task of the same generation (cancelled by a disconnect or stop). The started state is published once per leadership term, so a retry does not trigger another group event. It follows camel-master's `MasterConsumer` (CAMEL-24583, CAMEL-24626) in its locking and its lost-leadership checks, with two differences: camel-zookeeper-master has no `backOffDelay`/`backOffMaxAttempts` options, so the retry delay is fixed at camel-master's default of 5 seconds and the retries do not stop while this node is the master (camel-master gives up after 10 attempts and stays leader w
ithout consuming); and each attempt creates a new delegate consumer (a failed `start()` already stopped the previous one), where camel-master reuses one. Each failed attempt is logged at WARN with its stack trace.
The defects were found with a TLA+ model of the consumer (group thread, connection thread, stopping thread, each statement one step, `close()` waiting at most 5 s): "a master that received a leadership event consumes" is violated in 3 steps with one failed start; "no delegate runs on a stopped route or a node that is not the master" in 5 steps (event, connection lost, stopConsumer, create, start). With the fix all properties hold (up to 3 events, 2 failed starts, a disconnect and a stop, deadlock check on); a fix without the generation check, or without the retry, violates them. The model does not cover reconnecting as master after a disconnect.
As @oscerd pointed out on the ticket, the old code only got stuck when the failure came at or after the assignment of `delegate` (`updateState` or `startService(delegate)`, which is where most consumers connect); here the consumer is only published after a successful start, and every failure is retried. He also noted that `delegate` and `delegateService` were plain fields written by two threads: all writes and the start-path reads are now under the lock, and both fields are `volatile` because `doSuspend`/`doResume` read them without it.
Affected: main, 4.22.x and 4.18.x (same code; 4.14.x is end of life).
Tests: new `MasterConsumerLeadershipTest`, no ZooKeeper: a `ManagedGroupFactory` bound in the registry returns a fake group whose events the test fires, and the delegate endpoint fails a start or holds `createConsumer` on a latch (latches, Awaitility). Without the main-code change:
```
MasterConsumerLeadershipTest.testFailedStartIsRetried:73 ยป ConditionTimeout ... The master must start its consumer after a failed start ==> expected: <1> but was: <0> within 20 seconds.
MasterConsumerLeadershipTest.testDisconnectedWhileCreatingConsumer:93 A node that lost the leadership must not consume ==> expected: <0> but was: <1>
MasterConsumerLeadershipTest.testStoppedWhileCreatingConsumer:110 A stopped route must not consume ==> expected: <0> but was: <1>
```
With the change the camel-zookeeper-master unit tests pass (14). The integration tests need Docker and were not run.
# Target
- [x] I checked that the commit is targeting the correct branch (Camel 4 uses the `main` branch)
# Tracking
- [x] If this is a large change, bug fix, or code improvement, I checked there is a [JIRA issue](https://issues.apache.org/jira/browse/CAMEL) filed for the change (usually before you start working on it).
# Apache Camel coding standards and style
- [x] I checked that each commit in the pull request has a meaningful subject line and body.
- [ ] I have run `mvn clean install -DskipTests` locally from root folder and I have committed all auto-generated changes.
(I built and tested the affected modules, including the formatter and import-sort plugins. I did not run the full root build.)
# AI-assisted contributions
- [x] If this PR includes AI-generated code, commits have proper co-authorship attribution (e.g., `Co-authored-by` trailers) and the PR description identifies the AI tool used.
This PR was prepared with Claude Code (Claude Opus 5.5). The commit carries a `Co-Authored-By` trailer.
_Claude Code on behalf of allthingssecurity_
๐ค Generated with [Claude Code](https://claude.com/claude-code)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25315: camel-zookeeper-master - retry a failed start of the master consumer, and do not start it after the leadership was lost or the route stopped [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27391:
URL: https://github.com/apache/camel/pull/27391#issuecomment-6010342435
<!-- ci-tested-modules -->
:test_tube: **CI tested the following changed modules:**
- `components/camel-zookeeper-master`
:repeat: **1 test passed only after a retry on JDK 25** (1 retried attempt)
<details><summary>Recovered flaky tests on JDK 25 (1)</summary>
| Module | Test | Failed attempts | First failure |
| --- | --- | --- | --- |
| `dsl/camel-jbang/camel-jbang-plugin-tui` | `AiPanelAcpTest.retryResendsTheLastQuestionToTheAgent` | 1 | expected: <2> but was: <1> |
> :information_source: These tests did **not** fail the build. Surefire retried them and they passed.
> Retries are enabled project-wide by `surefire.rerunFailingTestsCount` in `parent/pom.xml`.
</details>
---
<details><summary>:microscope: Scalpel shadow comparison โ Scalpel: 9 of 698 tested, 27 compile-only โ current: 9 all tested</summary>
[Maveniverse Scalpel](https://github.com/maveniverse/scalpel) detected **9 affected modules** (current approach: 9).
**Skip-tests mode would test 9 modules** (1 direct + 8 downstream), **skip tests for 27** (generated code, meta-modules)
<details><summary>Modules Scalpel would test (9)</summary>
- `camel-jbang-mcp` โ downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-mcp` โ downstream of org.apache.camel:camel-jbang-core
- `camel-jbang-plugin-route-parser` โ downstream of org.apache.camel:camel-route-parser
- `camel-jbang-plugin-tui` โ downstream of org.apache.camel:camel-catalog
- `camel-jbang-plugin-validate` โ downstream of org.apache.camel:camel-yaml-dsl-validator
- `camel-launcher-container` โ downstream of org.apache.camel:camel-launcher
- `camel-yaml-dsl-validator` โ downstream of org.apache.camel:camel-catalog
- `camel-yaml-dsl-validator-maven-plugin` โ downstream of org.apache.camel:camel-yaml-dsl-validator
- `camel-zookeeper-master` โ components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java, components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterConsumerLeadershipTest.java
</details>
<details><summary>Modules with tests skipped (27)</summary>
- `apache-camel`
- `camel-allcomponents`
- `camel-catalog`
- `camel-catalog-console`
- `camel-catalog-maven`
- `camel-catalog-suggest`
- `camel-componentdsl`
- `camel-endpointdsl`
- `camel-endpointdsl-support`
- `camel-itest`
- `camel-jbang-core`
- `camel-jbang-it`
- `camel-jbang-main`
- `camel-jbang-plugin-edit`
- `camel-jbang-plugin-generate`
- `camel-jbang-plugin-kubernetes`
- `camel-jbang-plugin-test`
- `camel-kamelet-main`
- `camel-launcher`
- `camel-report-maven-plugin`
- `camel-route-parser`
- `camel-yaml-dsl`
- `camel-yaml-dsl-deserializers`
- `camel-yaml-dsl-maven-plugin`
- `coverage`
- `docs`
- `dummy-component`
</details>
> :information_source: Shadow mode โ Scalpel observes but does not affect test execution. [Learn more](https://github.com/maveniverse/scalpel)
</details>
<details><summary>All tested modules (36 modules, 5m 38s total)</summary>
**Total reactor time:** 5m 38s
| Module | Duration | Status |
| --- | --- | --- |
| Camel :: Launcher | 47.7s | SUCCESS |
| Camel :: JBang :: Plugin :: TUI | 43.1s | SUCCESS |
| Camel :: JBang :: MCP | 38.4s | SUCCESS |
| Camel :: Component DSL | 28.1s | SUCCESS |
| Camel :: Catalog :: Camel Catalog | 23.9s | SUCCESS |
| Camel :: YAML DSL | 19.6s | SUCCESS |
| Camel :: YAML DSL :: Validator | 19.2s | SUCCESS |
| Camel :: JBang :: Plugin :: Kubernetes | 17.7s | SUCCESS |
| Camel :: Docs | 15.7s | SUCCESS |
| Camel :: Kamelet Main | 12.9s | SUCCESS |
| Camel :: Catalog :: Camel Report Maven Plugin | 9.6s | SUCCESS |
| Camel :: YAML DSL :: Deserializers | 9.3s | SUCCESS |
| Camel :: JBang :: Plugin :: Testing | 8.3s | SUCCESS |
| Camel :: Catalog :: Camel Route Parser | 7.6s | SUCCESS |
| Camel :: All Components Sync point | 5.6s | SUCCESS |
| Camel :: JBang :: Plugin :: Validate | 5.2s | SUCCESS |
| Camel :: YAML DSL :: Validator Maven Plugin | 5.0s | SUCCESS |
| Camel :: YAML DSL :: Maven Plugins | 3.5s | SUCCESS |
| Camel :: Catalog :: Suggest (deprecated) | 2.4s | SUCCESS |
| Camel :: Catalog :: Maven | 2.2s | SUCCESS |
| Camel :: Assembly | 1.6s | SUCCESS |
| Camel :: Catalog :: Dummy Component | 1.6s | SUCCESS |
| Camel :: Coverage | 1.4s | SUCCESS |
| Camel :: JBang :: Plugin :: Edit | 1.4s | SUCCESS |
| Camel :: Catalog :: Console | 1.1s | SUCCESS |
| Camel :: JBang :: Integration tests | 1.1s | SUCCESS |
| Camel :: JBang :: Plugin :: Generate | 1.1s | SUCCESS |
| Camel :: JBang :: Plugin :: MCP | 0.8s | SUCCESS |
| Camel :: JBang :: Main | 0.7s | SUCCESS |
| Camel :: Launcher :: Container | 0.7s | SUCCESS |
| Camel :: JBang :: Plugin :: Route Parser | 0.6s | SUCCESS |
| Camel :: Endpoint DSL :: Support | 0.6s | SUCCESS |
| Camel :: Endpoint DSL | n/a | |
| Camel :: Integration Tests | n/a | |
| Camel :: JBang :: Core | n/a | |
| Camel :: Zookeeper Master | n/a | |
**Top 20 slowest modules:**
- `Camel :: Launcher` (47.7s)
- `Camel :: JBang :: Plugin :: TUI` (43.1s)
- `Camel :: JBang :: MCP` (38.4s)
- `Camel :: Component DSL` (28.1s)
- `Camel :: Catalog :: Camel Catalog` (23.9s)
- `Camel :: YAML DSL` (19.6s)
- `Camel :: YAML DSL :: Validator` (19.2s)
- `Camel :: JBang :: Plugin :: Kubernetes` (17.7s)
- `Camel :: Docs` (15.7s)
- `Camel :: Kamelet Main` (12.9s)
- `Camel :: Catalog :: Camel Report Maven Plugin` (9.6s)
- `Camel :: YAML DSL :: Deserializers` (9.3s)
- `Camel :: JBang :: Plugin :: Testing` (8.3s)
- `Camel :: Catalog :: Camel Route Parser` (7.6s)
- `Camel :: All Components Sync point` (5.6s)
- `Camel :: JBang :: Plugin :: Validate` (5.2s)
- `Camel :: YAML DSL :: Validator Maven Plugin` (5.0s)
- `Camel :: YAML DSL :: Maven Plugins` (3.5s)
- `Camel :: Catalog :: Suggest (deprecated)` (2.4s)
- `Camel :: Catalog :: Maven` (2.2s)
</details>
---
:gear: [View full build and test results](https://github.com/apache/camel/actions/runs/37392018868)
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25315: camel-zookeeper-master - retry a failed start of the master consumer, and do not start it after the leadership was lost or the route stopped [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus commented on code in PR #27391:
URL: https://github.com/apache/camel/pull/27391#discussion_r4192760189
##########
components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterConsumerLeadershipTest.java:
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.camel.component.zookeepermaster;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.GroupListener;
+import org.apache.camel.component.zookeepermaster.group.ManagedGroupFactory;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.curator.framework.CuratorFramework;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The leadership events of the group against the consumer of the master endpoint, with a fake group (no ZooKeeper): the
+ * CHANGED events are delivered on the group thread, DISCONNECTED on the Curator connection thread.
+ */
+class MasterConsumerLeadershipTest extends CamelTestSupport {
+
+ @BindToRegistry("fakeGroupFactory")
+ private final FakeGroupFactory groupFactory = new FakeGroupFactory();
+
+ private final ControlledEndpoint delegateEndpoint = new ControlledEndpoint();
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ void testFailedStartIsRetried() throws Exception {
+ startRoute();
+ masterConsumer().setRetryDelay(10);
+ delegateEndpoint.failStarts.set(1);
+
+ // elected: the start of the consumer fails (broker or server not available yet)
+ groupFactory.group.fireChanged();
+ assertEquals(0, delegateEndpoint.running.get());
+
+ // the master must try again, otherwise this node holds the leadership and nobody consumes
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, delegateEndpoint.running.get(),
+ "The master must start its consumer after a failed start"));
+ }
+
+ @Test
+ void testDisconnectedWhileCreatingConsumer() throws Exception {
+ startRoute();
+ delegateEndpoint.createGate = new CountDownLatch(1);
+
+ Thread groupThread = new Thread(groupFactory.group::fireChanged, "group");
+ groupThread.start();
+ assertTrue(delegateEndpoint.creating.await(20, TimeUnit.SECONDS));
+
+ // the connection to ZooKeeper is lost: this node is no longer the master
+ groupFactory.group.connected = false;
+ groupFactory.group.master = false;
+ groupFactory.group.fire(GroupListener.GroupEvent.DISCONNECTED);
+
+ delegateEndpoint.createGate.countDown();
+ groupThread.join(20000);
+ assertEquals(0, delegateEndpoint.running.get(), "A node that lost the leadership must not consume");
+ }
+
+ @Test
+ void testStoppedWhileCreatingConsumer() throws Exception {
+ startRoute();
+ delegateEndpoint.createGate = new CountDownLatch(1);
+
+ Thread groupThread = new Thread(groupFactory.group::fireChanged, "group");
+ groupThread.start();
+ assertTrue(delegateEndpoint.creating.await(20, TimeUnit.SECONDS));
+
+ // the start takes longer than the 5 seconds the group waits for its thread when it is closed
+ context.getRouteController().stopRoute("master");
+
+ delegateEndpoint.createGate.countDown();
+ groupThread.join(20000);
+ assertEquals(0, delegateEndpoint.running.get(), "A stopped route must not consume");
+ }
+
+ @Test
+ void testReconnectedWhileRetryIsStarting() throws Exception {
+ startRoute();
+ masterConsumer().setRetryDelay(10);
+ delegateEndpoint.failStarts.set(1);
+ // the first start fails, the retry (on the retry thread) is held while it creates its consumer
+ delegateEndpoint.gateFrom = 2;
+ delegateEndpoint.createGate = new CountDownLatch(1);
+
+ groupFactory.group.fireChanged();
+ assertTrue(delegateEndpoint.creating.await(20, TimeUnit.SECONDS));
+
+ // ZooKeeper suspends and reconnects meanwhile, and this node is elected again
+ groupFactory.group.connected = false;
+ groupFactory.group.master = false;
+ groupFactory.group.fire(GroupListener.GroupEvent.DISCONNECTED);
+ groupFactory.group.connected = true;
+ groupFactory.group.master = true;
+ groupFactory.group.fireChanged();
+
+ delegateEndpoint.createGate.countDown();
+ // the retry stops its consumer (it began before the disconnect), and the master must start one again
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, delegateEndpoint.running.get(),
+ "A master elected again while a start was in flight must consume"));
+ }
+
+ @Test
+ void testConsumerRestartedWhileCreatingConsumer() throws Exception {
+ startRoute();
+ CountDownLatch gate = new CountDownLatch(1);
+ delegateEndpoint.createGate = gate;
+
+ Thread groupThread = new Thread(groupFactory.group::fireChanged, "group");
+ groupThread.start();
+ assertTrue(delegateEndpoint.creating.await(20, TimeUnit.SECONDS));
+ // only that create is slow
+ delegateEndpoint.createGate = null;
Review Comment:
Nit: `createConsumer` reads `createGate` after `creating.countDown()`, so if the group thread reads it after this line it sees `null` and does not block. The test then still passes, but without exercising the race it is meant to cover. Reading the gate before `creating.countDown()` (or limiting the gate by a count instead of nulling it) makes it deterministic.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25315: camel-zookeeper-master - retry a failed start of the master consumer, and do not start it after the leadership was lost or the route stopped [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
allthingssecurity commented on code in PR #27391:
URL: https://github.com/apache/camel/pull/27391#discussion_r4192810312
##########
components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterConsumerLeadershipTest.java:
##########
@@ -0,0 +1,347 @@
+/*
+ * 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.camel.component.zookeepermaster;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.BindToRegistry;
+import org.apache.camel.Consumer;
+import org.apache.camel.Processor;
+import org.apache.camel.Producer;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.component.zookeepermaster.group.Group;
+import org.apache.camel.component.zookeepermaster.group.GroupListener;
+import org.apache.camel.component.zookeepermaster.group.ManagedGroupFactory;
+import org.apache.camel.component.zookeepermaster.group.NodeState;
+import org.apache.camel.support.DefaultConsumer;
+import org.apache.camel.support.DefaultEndpoint;
+import org.apache.camel.test.junit6.CamelTestSupport;
+import org.apache.curator.framework.CuratorFramework;
+import org.junit.jupiter.api.Test;
+
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * The leadership events of the group against the consumer of the master endpoint, with a fake group (no ZooKeeper): the
+ * CHANGED events are delivered on the group thread, DISCONNECTED on the Curator connection thread.
+ */
+class MasterConsumerLeadershipTest extends CamelTestSupport {
+
+ @BindToRegistry("fakeGroupFactory")
+ private final FakeGroupFactory groupFactory = new FakeGroupFactory();
+
+ private final ControlledEndpoint delegateEndpoint = new ControlledEndpoint();
+
+ @Override
+ public boolean isUseRouteBuilder() {
+ return false;
+ }
+
+ @Test
+ void testFailedStartIsRetried() throws Exception {
+ startRoute();
+ masterConsumer().setRetryDelay(10);
+ delegateEndpoint.failStarts.set(1);
+
+ // elected: the start of the consumer fails (broker or server not available yet)
+ groupFactory.group.fireChanged();
+ assertEquals(0, delegateEndpoint.running.get());
+
+ // the master must try again, otherwise this node holds the leadership and nobody consumes
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, delegateEndpoint.running.get(),
+ "The master must start its consumer after a failed start"));
+ }
+
+ @Test
+ void testDisconnectedWhileCreatingConsumer() throws Exception {
+ startRoute();
+ delegateEndpoint.createGate = new CountDownLatch(1);
+
+ Thread groupThread = new Thread(groupFactory.group::fireChanged, "group");
+ groupThread.start();
+ assertTrue(delegateEndpoint.creating.await(20, TimeUnit.SECONDS));
+
+ // the connection to ZooKeeper is lost: this node is no longer the master
+ groupFactory.group.connected = false;
+ groupFactory.group.master = false;
+ groupFactory.group.fire(GroupListener.GroupEvent.DISCONNECTED);
+
+ delegateEndpoint.createGate.countDown();
+ groupThread.join(20000);
+ assertEquals(0, delegateEndpoint.running.get(), "A node that lost the leadership must not consume");
+ }
+
+ @Test
+ void testStoppedWhileCreatingConsumer() throws Exception {
+ startRoute();
+ delegateEndpoint.createGate = new CountDownLatch(1);
+
+ Thread groupThread = new Thread(groupFactory.group::fireChanged, "group");
+ groupThread.start();
+ assertTrue(delegateEndpoint.creating.await(20, TimeUnit.SECONDS));
+
+ // the start takes longer than the 5 seconds the group waits for its thread when it is closed
+ context.getRouteController().stopRoute("master");
+
+ delegateEndpoint.createGate.countDown();
+ groupThread.join(20000);
+ assertEquals(0, delegateEndpoint.running.get(), "A stopped route must not consume");
+ }
+
+ @Test
+ void testReconnectedWhileRetryIsStarting() throws Exception {
+ startRoute();
+ masterConsumer().setRetryDelay(10);
+ delegateEndpoint.failStarts.set(1);
+ // the first start fails, the retry (on the retry thread) is held while it creates its consumer
+ delegateEndpoint.gateFrom = 2;
+ delegateEndpoint.createGate = new CountDownLatch(1);
+
+ groupFactory.group.fireChanged();
+ assertTrue(delegateEndpoint.creating.await(20, TimeUnit.SECONDS));
+
+ // ZooKeeper suspends and reconnects meanwhile, and this node is elected again
+ groupFactory.group.connected = false;
+ groupFactory.group.master = false;
+ groupFactory.group.fire(GroupListener.GroupEvent.DISCONNECTED);
+ groupFactory.group.connected = true;
+ groupFactory.group.master = true;
+ groupFactory.group.fireChanged();
+
+ delegateEndpoint.createGate.countDown();
+ // the retry stops its consumer (it began before the disconnect), and the master must start one again
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() -> assertEquals(1, delegateEndpoint.running.get(),
+ "A master elected again while a start was in flight must consume"));
+ }
+
+ @Test
+ void testConsumerRestartedWhileCreatingConsumer() throws Exception {
+ startRoute();
+ CountDownLatch gate = new CountDownLatch(1);
+ delegateEndpoint.createGate = gate;
+
+ Thread groupThread = new Thread(groupFactory.group::fireChanged, "group");
+ groupThread.start();
+ assertTrue(delegateEndpoint.creating.await(20, TimeUnit.SECONDS));
+ // only that create is slow
+ delegateEndpoint.createGate = null;
Review Comment:
Good catch, thanks. Fixed in 0588aca87d65: `createConsumer` now reads the gate before `creating.countDown()`, so the held create always blocks. Also took the optional point: `retryExecutor` and `groupListener` are volatile now. Module tests: 16 pass.
_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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25315: camel-zookeeper-master - retry a failed start of the master consumer, and do not start it after the leadership was lost or the route stopped [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25315: camel-zookeeper-master - retry a failed start of the master consumer, and do not start it after the leadership was lost or the route stopped [camel]
Posted by "allthingssecurity (via GitHub)" <gi...@apache.org>.
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25315: camel-zookeeper-master - retry a failed start of the master consumer, and do not start it after the leadership was lost or the route stopped [camel]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #27391:
URL: https://github.com/apache/camel/pull/27391#issuecomment-6010013084
:star2: Thank you for your contribution to the Apache Camel project! :star2:
:robot: CI automation will test this PR automatically.
:camel: Apache Camel Committers, please review the following items:
* First-time contributors **require MANUAL approval** for the GitHub Actions to run
* You can use the command `/component-test (camel-)component-name1 (camel-)component-name2..` to request a test from the test bot although they are normally detected and executed by CI.
* You can label PRs using `skip-tests` and `test-dependents` to fine-tune the checks executed by this PR.
* Build and test logs are available in the summary page. **Only** [Apache Camel committers](https://camel.apache.org/community/team/#committers) have access to the summary.
:warning: Be careful when sharing logs. Review their contents before sharing them publicly.
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] CAMEL-25315: camel-zookeeper-master - retry a failed start of the master consumer, and do not start it after the leadership was lost or the route stopped [camel]
Posted by "davsclaus (via GitHub)" <gi...@apache.org>.
davsclaus merged PR #27391:
URL: https://github.com/apache/camel/pull/27391
--
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: commits-unsubscribe@camel.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org