This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new e670025a49a4 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 (#27391)
e670025a49a4 is described below
commit e670025a49a4aeeffb4b4f8b81a99c7e2bf78646
Author: allthingssecurity <[email protected]>
AuthorDate: Tue Oct 6 14:52:24 2026 +0530
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 (#27391)
* 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
MasterConsumer starts the delegate consumer from the CHANGED event of the
group thread (onLockOwned) and stops it from DISCONNECTED (the Curator
connection thread, and the closing thread when the group is closed) and from
doStop, without any synchronisation, and only if the delegate field is null.
- A delegate whose start failed (the broker or server is not reachable when
the node is elected) stayed in the field and was never started again: the
node kept the leadership and nobody consumed until the connection was lost
or the route was restarted.
- A disconnect (or a stop) between the null check and the assignment of the
new delegate stopped nothing, and the delegate was then created and
started:
it kept consuming on a node that was no longer the master (also after it
reconnected as a standby, as standby events stop nothing), or after the
route was stopped (when the start took longer than the 5 seconds the group
waits for its thread on close).
The consumer now guards the delegate with a lock (like the MasterConsumer of
camel-master): a start is marked under the lock (after checking again that
the group is connected and this node is the master), the delegate is created
and started without holding it, and it is published under the lock only if
no
disconnect or stop happened meanwhile (a generation counter), otherwise it
is
stopped. A failed start is retried after 5 seconds while this node is still
the master; the started state is published to the group once per leadership
term, so that a retry does not trigger another group event.
delegate and delegateService are also volatile: doSuspend/doResume read
them without the lock.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25315: camel-zookeeper-master - start the consumer again when
elected while a start was in flight
A leadership event that arrives while a start is in flight is skipped. When
that start then ends after a disconnect
(or a stop of the consumer), it stopped its consumer and scheduled nothing,
so a node elected again meanwhile held the
leadership with no consumer: a reconnect while a retry runs on the retry
thread, or a stop and start of the same
consumer instance while the old group thread is still in a slow start. Such
a start now schedules a start for the
current generation without delay when the node is the leader, runs and has
no consumer or scheduled start.
Review follow-ups: a failed start goes to getExceptionHandler() (as
camel-master), with the attempt number; a
package-private setRetryDelay for the tests; a stale retry task no longer
clears the reference of a newer one.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* CAMEL-25315: camel-zookeeper-master - Read the test gate before
signalling, and make the retry executor and group listener volatile
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../component/zookeepermaster/MasterConsumer.java | 188 +++++++++--
.../MasterConsumerLeadershipTest.java | 348 +++++++++++++++++++++
2 files changed, 509 insertions(+), 27 deletions(-)
diff --git
a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java
b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java
index a35136ab889e..8a42c3bdf347 100644
---
a/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java
+++
b/components/camel-zookeeper-master/src/main/java/org/apache/camel/component/zookeepermaster/MasterConsumer.java
@@ -16,6 +16,12 @@
*/
package org.apache.camel.component.zookeepermaster;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+
import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
@@ -36,13 +42,31 @@ import org.slf4j.LoggerFactory;
@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;
- private ZookeeperGroupListenerSupport groupListener;
+ private volatile ZookeeperGroupListenerSupport groupListener;
private final MasterEndpoint endpoint;
private final Processor processor;
- private Consumer delegate;
- private SuspendableService delegateService;
+ private volatile Consumer delegate;
+ private volatile SuspendableService delegateService;
private volatile CamelNodeState thisNodeState;
+ // the leadership events come from the group thread (CHANGED) and the
ZooKeeper connection thread
+ // (DISCONNECTED), and the consumer can be stopped meanwhile: the delegate
and these fields are guarded by the lock
+ // (delegate and delegateService are also volatile, as doSuspend/doResume
read them without it)
+ private final Lock leadershipLock = new ReentrantLock();
+ // incremented when the leadership may be lost or the consumer stops: a
start that began before does not publish
+ // its consumer, it stops it
+ private long generation;
+ private boolean starting;
+ // the state published when this node starts its consumer, once per
leadership term (a new state is a change of
+ // the group, which would trigger another leadership event)
+ private CamelNodeState startedState;
+ private volatile ScheduledExecutorService retryExecutor;
+ private ScheduledFuture<?> retryTask;
+ private volatile long retryDelay = RETRY_DELAY_MILLIS;
+ // failed starts in a row, for the log
+ private int failedStarts;
public MasterConsumer(MasterEndpoint endpoint, Processor processor) {
super(endpoint, processor);
@@ -94,6 +118,8 @@ public class MasterConsumer extends DefaultConsumer {
this.groupListener.setZooKeeperPassword(endpoint.getComponent().getZooKeeperPassword());
this.groupListener.setCurator(endpoint.getComponent().getCurator());
this.groupListener.setMaximumConnectionTimeout(endpoint.getComponent().getMaximumConnectionTimeout());
+ this.retryExecutor =
endpoint.getCamelContext().getExecutorServiceManager()
+ .newSingleThreadScheduledExecutor(this,
"ZooKeeperMasterRetry");
ServiceHelper.startService(groupListener);
LOG.info("Attempting to become master for endpoint: {} in {} with
singletonID: {}", endpoint,
@@ -105,9 +131,21 @@ public class MasterConsumer extends DefaultConsumer {
@Override
protected void doStop() throws Exception {
try {
- stopConsumer();
+ leadershipLock.lock();
+ try {
+ generation++;
+ failedStarts = 0;
+ cancelRetry();
+ stopConsumer();
+ } finally {
+ leadershipLock.unlock();
+ }
} finally {
ServiceHelper.stopAndShutdownServices(groupListener);
+ if (retryExecutor != null) {
+
endpoint.getCamelContext().getExecutorServiceManager().shutdownNow(retryExecutor);
+ retryExecutor = null;
+ }
}
super.doStop();
}
@@ -125,6 +163,21 @@ public class MasterConsumer extends DefaultConsumer {
delegate = null;
delegateService = null;
thisNodeState = null;
+ startedState = null;
+ }
+
+ /**
+ * The delay before the master tries again to start a consumer whose start
failed (5 seconds). For tests.
+ */
+ void setRetryDelay(long retryDelay) {
+ this.retryDelay = retryDelay;
+ }
+
+ private void cancelRetry() {
+ if (retryTask != null) {
+ retryTask.cancel(false);
+ retryTask = null;
+ }
}
@Override
@@ -144,40 +197,121 @@ public class MasterConsumer extends DefaultConsumer {
}
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 && retryGeneration == generation) {
+ // this is the scheduled task (a task of an older generation
was cancelled, and a newer one may be
+ // scheduled meanwhile)
+ 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);
+ // a leadership event that came meanwhile (reconnected, or the
route was started again) was skipped,
+ // as this start was in flight: start again if this node is
the master now
+ if (isRunAllowed() && isLeader() && delegate == null &&
retryTask == null) {
+ LOG.info("Elected as master again while the consumer
started. Starting consumer again: {}",
+ endpoint.getConsumerEndpoint());
+ scheduleStart(0);
+ }
+ } else if (cause != null) {
+ // forget the consumer (its start stopped it) and try again
later, as long as this node is the master
+ failedStarts++;
+ getExceptionHandler().handleException("Failed to start master
consumer for: " + endpoint + " (attempt #"
+ + failedStarts + ").
Trying again in " + retryDelay + " millis.",
+ cause);
+ scheduleStart(retryDelay);
+ } else {
+ failedStarts = 0;
+ delegate = consumer;
+ delegateService = consumer instanceof SuspendableService
suspendable ? suspendable : null;
LOG.info("Elected as master. Consumer started: {}",
endpoint.getConsumerEndpoint());
}
- };
+ } finally {
+ leadershipLock.unlock();
+ }
+ }
+
+ // under the lock: starts the consumer for the current generation, unless
a disconnect or a stop comes first
+ private void scheduleStart(long delay) {
+ if (retryExecutor != null) {
+ final long scheduledGeneration = generation;
+ retryTask = retryExecutor.schedule(() ->
startConsumer(scheduledGeneration), delay, TimeUnit.MILLISECONDS);
+ }
+ }
+
+ private boolean isLeader() {
+ // the group sets connected to false before it tells the listeners
that it is disconnected
+ return groupListener != null && groupListener.getGroup() != null &&
groupListener.getGroup().isConnected()
+ && groupListener.getGroup().isMaster();
}
protected Runnable onDisconnected() {
return () -> {
+ leadershipLock.lock();
try {
+ generation++;
+ failedStarts = 0;
+ cancelRetry();
stopConsumer();
} catch (Exception e) {
LOG.warn("Failed to stop master consumer for: {}", endpoint,
e);
+ } finally {
+ leadershipLock.unlock();
}
};
}
diff --git
a/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterConsumerLeadershipTest.java
b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterConsumerLeadershipTest.java
new file mode 100644
index 000000000000..fe81be182d2f
--- /dev/null
+++
b/components/camel-zookeeper-master/src/test/java/org/apache/camel/component/zookeepermaster/MasterConsumerLeadershipTest.java
@@ -0,0 +1,348 @@
+/*
+ * 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;
+
+ // the consumer is stopped and started again (as by JMX; a route
restart creates a new consumer) while the old
+ // group thread is still in the slow start, and the new group elects
this node
+ MasterConsumer consumer = masterConsumer();
+ consumer.stop();
+ consumer.start();
+ groupFactory.group.fireChanged();
+
+ gate.countDown();
+ groupThread.join(20000);
+ await().atMost(20, TimeUnit.SECONDS).untilAsserted(() ->
assertEquals(1, delegateEndpoint.running.get(),
+ "A master started again while its old start was in flight must
consume"));
+ }
+
+ private MasterConsumer masterConsumer() {
+ return (MasterConsumer) context.getRoute("master").getConsumer();
+ }
+
+ private void startRoute() throws Exception {
+ context.addEndpoint("controlled:delegate", delegateEndpoint);
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+
from("zookeeper-master:test:controlled:delegate").routeId("master").to("mock:result");
+ }
+ });
+ context.start();
+ assertTrue(groupFactory.group != null &&
groupFactory.group.listeners.size() == 1);
+ }
+
+ private final class ControlledEndpoint extends DefaultEndpoint {
+ final AtomicInteger failStarts = new AtomicInteger();
+ final AtomicInteger running = new AtomicInteger();
+ final AtomicInteger creates = new AtomicInteger();
+ final CountDownLatch creating = new CountDownLatch(1);
+ // the createGate holds the creates from this one on (1 = the first)
+ volatile int gateFrom = 1;
+ volatile CountDownLatch createGate;
+
+ @Override
+ protected String createEndpointUri() {
+ return "controlled:delegate";
+ }
+
+ @Override
+ public Producer createProducer() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Consumer createConsumer(Processor processor) throws Exception {
+ if (creates.incrementAndGet() < gateFrom) {
+ return newConsumer(processor);
+ }
+ // read the gate before signalling, as the test may clear it right
after that signal
+ CountDownLatch gate = createGate;
+ creating.countDown();
+ if (gate != null) {
+ assertTrue(gate.await(20, TimeUnit.SECONDS));
+ }
+ return newConsumer(processor);
+ }
+
+ private Consumer newConsumer(Processor processor) {
+ return new DefaultConsumer(this, processor) {
+ private boolean up;
+
+ @Override
+ protected void doStart() throws Exception {
+ if (failStarts.getAndDecrement() > 0) {
+ throw new IllegalStateException("Simulated start
failure");
+ }
+ super.doStart();
+ up = true;
+ running.incrementAndGet();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ if (up) {
+ up = false;
+ running.decrementAndGet();
+ }
+ super.doStop();
+ }
+ };
+ }
+ }
+
+ private static final class FakeGroupFactory implements ManagedGroupFactory
{
+ volatile FakeGroup group;
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public <T extends NodeState> Group<T> createGroup(String path,
Class<T> clazz) {
+ group = new FakeGroup();
+ return (Group<T>) group;
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createGroup(String path,
Class<T> clazz, ThreadFactory threadFactory) {
+ return createGroup(path, clazz);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path,
Class<T> clazz) {
+ return createGroup(path, clazz);
+ }
+
+ @Override
+ public <T extends NodeState> Group<T> createMultiGroup(String path,
Class<T> clazz, ThreadFactory threadFactory) {
+ return createGroup(path, clazz);
+ }
+
+ @Override
+ public CuratorFramework getCurator() {
+ return null;
+ }
+
+ @Override
+ public void close() {
+ // noop
+ }
+ }
+
+ private static final class FakeGroup implements Group<CamelNodeState> {
+ final List<GroupListener<CamelNodeState>> listeners = new
CopyOnWriteArrayList<>();
+ volatile boolean connected = true;
+ volatile boolean master = true;
+
+ void fireChanged() {
+ fire(GroupListener.GroupEvent.CHANGED);
+ }
+
+ void fire(GroupListener.GroupEvent event) {
+ listeners.forEach(l -> l.groupEvent(this, event));
+ }
+
+ @Override
+ public boolean isConnected() {
+ return connected;
+ }
+
+ @Override
+ public void start() {
+ // noop
+ }
+
+ @Override
+ public void close() {
+ // as ZooKeeperGroup: a group closed while connected tells its
listeners it is disconnected
+ if (connected) {
+ fire(GroupListener.GroupEvent.DISCONNECTED);
+ }
+ listeners.clear();
+ }
+
+ @Override
+ public void add(GroupListener<CamelNodeState> listener) {
+ listeners.add(listener);
+ }
+
+ @Override
+ public void remove(GroupListener<CamelNodeState> listener) {
+ listeners.remove(listener);
+ }
+
+ @Override
+ public void update(CamelNodeState state) {
+ // noop
+ }
+
+ @Override
+ public Map<String, CamelNodeState> members() {
+ return Collections.emptyMap();
+ }
+
+ @Override
+ public boolean isMaster() {
+ return master;
+ }
+
+ @Override
+ public CamelNodeState master() {
+ return null;
+ }
+
+ @Override
+ public List<CamelNodeState> slaves() {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public CamelNodeState getLastState() {
+ return null;
+ }
+ }
+}
