This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new eb8b80400e [Fix][Zeta] Fix lite member (worker) cannot join cluster
when started before master (#12755)
eb8b80400e is described below
commit eb8b80400ece6b81525529c4c2ab634839f38533
Author: WPNG <[email protected]>
AuthorDate: Tue Oct 27 12:37:18 2026 +0000
[Fix][Zeta] Fix lite member (worker) cannot join cluster when started
before master (#12755)
Co-authored-by: wpngwang <[email protected]>
---
.../engines/zeta/separated-cluster-deployment.md | 4 +
.../engines/zeta/separated-cluster-deployment.md | 2 +
.../server/joiner/LiteNodeDropOutTcpIpJoiner.java | 146 +++++-
.../joiner/LiteNodeDropOutDiscoveryJoinerTest.java | 504 +++++++++++++++++++++
.../joiner/LiteNodeDropOutTcpIpJoinerTest.java | 382 ++++++++++++++++
5 files changed, 1032 insertions(+), 6 deletions(-)
diff --git a/docs/en/engines/zeta/separated-cluster-deployment.md
b/docs/en/engines/zeta/separated-cluster-deployment.md
index 6eee00d7c7..bcc20d1393 100644
--- a/docs/en/engines/zeta/separated-cluster-deployment.md
+++ b/docs/en/engines/zeta/separated-cluster-deployment.md
@@ -587,6 +587,10 @@ mkdir -p $SEATUNNEL_HOME/logs
The logs will be written to `$SEATUNNEL_HOME/logs/seatunnel-engine-worker.log`.
+> A Worker started before any Master will keep retrying the configured master
addresses for up to
+> `hazelcast.max.join.seconds` (default 300s) and log `This lite member cannot
assume the master
+> role ...` while waiting.
+
## 8. Submit And Manage Jobs
### 8.1 Submit Jobs With The SeaTunnel Engine Client
diff --git a/docs/zh/engines/zeta/separated-cluster-deployment.md
b/docs/zh/engines/zeta/separated-cluster-deployment.md
index ed0300d5cf..311b0d4f01 100644
--- a/docs/zh/engines/zeta/separated-cluster-deployment.md
+++ b/docs/zh/engines/zeta/separated-cluster-deployment.md
@@ -546,6 +546,8 @@ mkdir -p $SEATUNNEL_HOME/logs
日志将写入 `$SEATUNNEL_HOME/logs/seatunnel-engine-worker.log`
+> 如果 Worker 先于任何 Master 启动，它会在 `hazelcast.max.join.seconds`（默认 300 秒）内持续重试配置的
master 地址，并在等待期间输出 `This lite member cannot assume the master role ...` 日志。
+
## 8. 安装 SeaTunnel Engine 客户端
### 8.1 和服务端一样设置`SEATUNNEL_HOME`
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutTcpIpJoiner.java
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutTcpIpJoiner.java
index afb3eab795..800da87b5c 100644
---
a/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutTcpIpJoiner.java
+++
b/seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutTcpIpJoiner.java
@@ -81,6 +81,49 @@ public class LiteNodeDropOutTcpIpJoiner extends TcpIpJoiner {
return claimingMastership;
}
+ /**
+ * Re-declared from {@link com.hazelcast.internal.cluster.Joiner} to
downgrade Hazelcast's INFO
+ * "added to the blacklist" log to FINE only for the noisy case: a lite
member temporarily
+ * blacklisting a candidate during the retry loop (it
blacklists/unblacklists every candidate
+ * each cycle). Non-lite members and permanent entries keep Hazelcast's
INFO so operators still
+ * see those signals; the one-shot state log in {@link #logRetryState}
already summarizes the
+ * lite-member retry.
+ *
+ * <p>The map bookkeeping mirrors Hazelcast 5.1 {@code
AbstractJoiner#blacklist} and {@code
+ * AbstractJoiner#unblacklist}; re-check these two methods when the shaded
Hazelcast is
+ * upgraded.
+ */
+ @Override
+ public void blacklist(Address address, boolean permanent) {
+ if (node.isLiteMember() && !permanent) {
+ Boolean previous = blacklistedAddresses.putIfAbsent(address,
permanent);
+ if (previous == null && logger.isFineEnabled()) {
+ logger.fine(address + " is added to the blacklist.");
+ }
+ return;
+ }
+ super.blacklist(address, permanent);
+ }
+
+ /**
+ * Re-declared from {@link com.hazelcast.internal.cluster.Joiner} to
downgrade Hazelcast's INFO
+ * "removed from the blacklist" log to FINE for lite members. See {@link
#blacklist(Address,
+ * boolean)}.
+ */
+ @Override
+ public boolean unblacklist(Address address) {
+ if (node.isLiteMember()) {
+ if (blacklistedAddresses.remove(address, Boolean.FALSE)) {
+ if (logger.isFineEnabled()) {
+ logger.fine(address + " is removed from the blacklist.");
+ }
+ return true;
+ }
+ return false;
+ }
+ return super.unblacklist(address);
+ }
+
private int getConnTimeoutSeconds() {
return joinConfig.getTcpIpConfig().getConnectionTimeoutSeconds();
}
@@ -146,8 +189,35 @@ public class LiteNodeDropOutTcpIpJoiner extends
TcpIpJoiner {
long maxJoinMillis = getMaxJoinMillis();
long startTime = Clock.currentTimeMillis();
+ // Tracks whether the retry state (no candidates / permanently
blacklisted /
+ // temporarily blacklisted) has been logged once for this join
attempt. Local to this
+ // method so that a re-join (for example after hot restart) logs
again.
+ boolean retryStateLogged = false;
+
+ // An interrupt must also terminate the retry attempt: with an
empty or fully
+ // permanently-blacklisted candidate list, tryJoinAddresses()
returns before reaching
+ // its own blockOnJoin(), so the retry helper would otherwise keep
throwing
+ // InterruptedException and spin until the join deadline. Checking
the flag here gives
+ // the loop an explicit exit while the helper restores the flag.
+ while (shouldRetry()
+ && !Thread.currentThread().isInterrupted()
+ && (Clock.currentTimeMillis() - startTime <
maxJoinMillis)) {
+ // If the initial join produced no candidates (for example a
discovery strategy
+ // that returned nothing before the master was published),
refresh once per pass so
+ // a master that appears later can still be found within the
overall join deadline.
+ // For TCP/IP this re-resolves a static member list and is
effectively a no-op.
+ if (possibleAddresses.isEmpty()) {
+ try {
+ possibleAddresses = getPossibleAddresses();
+ } catch (Exception e) {
+ // A transient discovery failure should not abort the
whole join attempt;
+ // keep the empty list so the next pass retries.
Errors (e.g. OOM) are left
+ // to propagate so a fatal condition is not swallowed
for the whole window.
+ logger.warning(
+ "Failed to refresh candidate master addresses;
will retry.", e);
+ }
+ }
- while (shouldRetry() && (Clock.currentTimeMillis() - startTime <
maxJoinMillis)) {
tryJoinAddresses(possibleAddresses);
if (clusterService.isJoined()) {
@@ -155,11 +225,27 @@ public class LiteNodeDropOutTcpIpJoiner extends
TcpIpJoiner {
}
// update for seatunnel, lite member can not become master node
- if (isAllBlacklisted(possibleAddresses) &&
!node.isLiteMember()) {
- logger.fine(
- "This node will assume master role since none of
the possible members accepted join request.");
- clusterJoinManager.setThisMemberAsMaster();
- return;
+ if (isAllBlacklisted(possibleAddresses)) {
+ if (node.isLiteMember()) {
+ boolean clearedAny =
retryTemporaryBlacklistedAddresses(possibleAddresses);
+ if (!retryStateLogged) {
+ retryStateLogged = true;
+ logRetryState(possibleAddresses, clearedAny);
+ }
+ if (!clusterService.isJoined()) {
+ // Reset any stale master address learned earlier
so the next pass
+ // re-asks all candidates instead of pinning to a
dead master.
+ clusterService.setMasterAddressToJoin(null);
+ }
+ // A lite member can never claim mastership, so skip
the redundant claim
+ // attempt and the master-address reset below.
+ continue;
+ } else {
+ logger.fine(
+ "This node will assume master role since none
of the possible members accepted join request.");
+ clusterJoinManager.setThisMemberAsMaster();
+ return;
+ }
}
if (tryClaimMastership(possibleAddresses)) {
@@ -173,6 +259,54 @@ public class LiteNodeDropOutTcpIpJoiner extends
TcpIpJoiner {
}
}
+ /**
+ * Waits for one join retry interval and clears only the temporary
blacklist entries so that the
+ * next {@link #tryJoinAddresses(Collection)} call retries them. Returns
{@code true} if at
+ * least one temporary entry was cleared, {@code false} if there was
nothing to clear (an empty
+ * candidate list, or every candidate is permanently blacklisted).
+ *
+ * <p>Permanently blacklisted addresses are intentionally left untouched:
{@link
+ * #unblacklist(Address)} only removes temporary (non-permanent) entries.
+ */
+ private boolean retryTemporaryBlacklistedAddresses(Collection<Address>
possibleAddresses) {
+ try {
+ clusterService.blockOnJoin(JOIN_RETRY_WAIT_TIME);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ boolean clearedAny = false;
+ for (Address address : possibleAddresses) {
+ if (unblacklist(address)) {
+ clearedAny = true;
+ }
+ }
+ return clearedAny;
+ }
+
+ /**
+ * Logs a single, accurate message describing why this lite member cannot
join yet, based on
+ * whether the candidate list is empty, every candidate is permanently
blacklisted, or there are
+ * temporarily blacklisted candidates that will keep being retried.
+ */
+ private void logRetryState(Collection<Address> possibleAddresses, boolean
clearedAny) {
+ if (possibleAddresses.isEmpty()) {
+ logger.warning(
+ "This lite member has no candidate master addresses to
join; it will keep "
+ + "re-checking for candidate master addresses
until a master is found or "
+ + "hazelcast.max.join.seconds elapses.");
+ } else if (!clearedAny) {
+ logger.warning(
+ "All candidate master addresses are permanently
blacklisted: "
+ + possibleAddresses
+ + "; this lite member cannot retry them.");
+ } else {
+ logger.info(
+ "This lite member cannot assume the master role and all
candidate master "
+ + "addresses are temporarily blacklisted. It will
keep retrying until a "
+ + "master becomes reachable or
hazelcast.max.join.seconds elapses.");
+ }
+ }
+
private boolean tryClaimMastership(Collection<Address> addresses) {
boolean consensus = false;
if (isThisNodeMasterCandidate(addresses)) {
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutDiscoveryJoinerTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutDiscoveryJoinerTest.java
new file mode 100644
index 0000000000..6d33176b1e
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutDiscoveryJoinerTest.java
@@ -0,0 +1,504 @@
+/*
+ * 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.seatunnel.engine.server.joiner;
+
+import org.apache.seatunnel.engine.server.TestUtils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.cluster.Address;
+import com.hazelcast.config.Config;
+import com.hazelcast.config.DiscoveryConfig;
+import com.hazelcast.config.DiscoveryStrategyConfig;
+import com.hazelcast.config.TcpIpConfig;
+import com.hazelcast.config.properties.PropertyDefinition;
+import com.hazelcast.instance.impl.DefaultNodeContext;
+import com.hazelcast.instance.impl.HazelcastInstanceFactory;
+import com.hazelcast.instance.impl.HazelcastInstanceImpl;
+import com.hazelcast.instance.impl.HazelcastInstanceProxy;
+import com.hazelcast.instance.impl.Node;
+import com.hazelcast.internal.cluster.Joiner;
+import com.hazelcast.logging.ILogger;
+import com.hazelcast.spi.discovery.DiscoveryNode;
+import com.hazelcast.spi.discovery.DiscoveryStrategy;
+import com.hazelcast.spi.discovery.DiscoveryStrategyFactory;
+import com.hazelcast.spi.discovery.SimpleDiscoveryNode;
+
+import java.net.UnknownHostException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+/**
+ * Verifies that the discovery-based lite-member joiner ({@link
LiteNodeDropOutDiscoveryJoiner})
+ * inherits the TCP/IP lite-member retry fix from {@link
LiteNodeDropOutTcpIpJoiner}, including
+ * refreshing an initially-empty discovery result within the overall join
deadline.
+ */
+class LiteNodeDropOutDiscoveryJoinerTest {
+
+ private static final String HOST = "127.0.0.1";
+
+ /**
+ * A discovery-based worker (lite member) starts before the master. The
discovery strategy
+ * already returns the master address before the master port is ready, so
the worker temporarily
+ * blacklists it and must keep retrying (inherited from the TCP/IP fix)
until the master starts
+ * and the worker joins.
+ */
+ @Test
+ void discoveryWorkerStartedBeforeMasterShouldEventuallyJoin() throws
Exception {
+ String clusterName =
+
TestUtils.getClusterName("LiteNodeDropOutDiscoveryJoinerTest_workerJoin");
+ int masterPort = TestUtils.getAvailablePort();
+ int workerPort = TestUtils.getAvailablePort();
+
+ Config workerConfig = newDiscoveryConfig(clusterName, workerPort,
masterPort);
+ workerConfig.setLiteMember(true);
+ workerConfig.setProperty("hazelcast.max.join.seconds", "60");
+ // Keep the initial discovery poll window short; the strategy returns
a node immediately.
+ workerConfig.setProperty("hazelcast.wait.seconds.before.join", "1");
+
+ // Fires once the worker has actually blacklisted the not-yet-started
master address.
+ CountDownLatch blacklistLatch = new CountDownLatch(1);
+
+ AtomicReference<HazelcastInstanceImpl> workerRef = new
AtomicReference<>();
+ AtomicReference<Throwable> workerError = new AtomicReference<>();
+ Thread workerThread =
+ new Thread(
+ () -> {
+ try {
+ workerRef.set(startNode(workerConfig,
blacklistLatch));
+ } catch (Throwable t) {
+ workerError.set(t);
+ }
+ },
+ "test-discovery-worker-started-before-master");
+ workerThread.start();
+
+ HazelcastInstanceImpl master = null;
+ try {
+ // Wait until the worker has provably blacklisted the
not-yet-started master address.
+ Assertions.assertTrue(
+ blacklistLatch.await(30, SECONDS),
+ "worker should have blacklisted the not-yet-started master
address");
+
+ Config masterConfig = newTcpIpConfig(clusterName, masterPort);
+ master = startNode(masterConfig, null);
+
+ workerThread.join(SECONDS.toMillis(60));
+ Assertions.assertFalse(workerThread.isAlive(), "worker should have
finished joining");
+ Assertions.assertNull(workerError.get(), "worker should not fail
to start");
+
+ HazelcastInstanceImpl worker = workerRef.get();
+ Assertions.assertNotNull(worker);
+ Assertions.assertTrue(worker.node.getClusterService().isJoined());
+ Assertions.assertFalse(worker.node.isMaster(), "a worker must
never become master");
+ Assertions.assertTrue(master.node.isMaster());
+ } finally {
+ workerThread.interrupt();
+ try {
+ workerThread.join(SECONDS.toMillis(5));
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ }
+ if (workerRef.get() != null) {
+ workerRef.get().shutdown();
+ }
+ if (master != null) {
+ master.shutdown();
+ }
+ // Force-stop any still-starting worker node (e.g. when a timeout
assertion failed
+ // before workerRef was set) so it cannot leak into later tests in
the same JVM.
+ HazelcastInstanceFactory.shutdownAll();
+ }
+ }
+
+ /**
+ * A discovery-based worker (lite member) starts before the master, but
discovery returns no
+ * candidate until after the initial discovery window. The worker must
keep refreshing discovery
+ * within the join deadline and join once the master becomes discoverable,
without ever becoming
+ * master itself.
+ *
+ * <p>The master is made discoverable only after the initial discovery
window ({@code
+ * hazelcast.wait.seconds.before.join}=1s) has elapsed, so the empty-list
refresh path (not the
+ * initial-discovery path) is exercised.
+ */
+ @Test
+ void discoveryWorkerStartedWithEmptyDiscoveryShouldEventuallyJoin() throws
Exception {
+ String clusterName =
+
TestUtils.getClusterName("LiteNodeDropOutDiscoveryJoinerTest_emptyJoin");
+ int masterPort = TestUtils.getAvailablePort();
+ int workerPort = TestUtils.getAvailablePort();
+
+ // Discovery returns no candidates until this latch is counted down
(i.e. the master is up).
+ CountDownLatch masterReady = new CountDownLatch(1);
+ AtomicInteger discoverCount = new AtomicInteger();
+
+ Config workerConfig =
+ newEmptyDiscoveryConfig(
+ clusterName, workerPort, masterPort, masterReady,
discoverCount);
+ workerConfig.setLiteMember(true);
+ workerConfig.setProperty("hazelcast.max.join.seconds", "60");
+ workerConfig.setProperty("hazelcast.wait.seconds.before.join", "1");
+
+ AtomicReference<HazelcastInstanceImpl> workerRef = new
AtomicReference<>();
+ AtomicReference<Throwable> workerError = new AtomicReference<>();
+ Thread workerThread =
+ new Thread(
+ () -> {
+ try {
+ workerRef.set(startNode(workerConfig, null));
+ } catch (Throwable t) {
+ workerError.set(t);
+ }
+ },
+ "test-discovery-empty-join");
+ workerThread.start();
+
+ HazelcastInstanceImpl master = null;
+ try {
+ // Wait until the worker has polled discovery enough times to be
past the initial
+ // discovery window (hazelcast.wait.seconds.before.join=1s) and
into the empty-list
+ // refresh loop. The initial window polls roughly 7 times with the
backoff idle
+ // strategy, so 10 polls guarantee the refresh path is exercised.
Polling the counter
+ // instead of a fixed sleep keeps this deterministic even on slow
runners.
+ long pollDeadline = System.currentTimeMillis() +
SECONDS.toMillis(30);
+ while (discoverCount.get() < 10 && System.currentTimeMillis() <
pollDeadline) {
+ Thread.sleep(100);
+ }
+ Assertions.assertTrue(
+ discoverCount.get() >= 10,
+ "worker should have polled discovery past the initial
window");
+
+ // Record how many times discovery has been polled so far, then
make the master
+ // discoverable and verify the worker polls discovery again (not
just joins).
+ int discoveriesBeforeMaster = discoverCount.get();
+
+ master = startNode(newTcpIpConfig(clusterName, masterPort), null);
+ masterReady.countDown();
+
+ workerThread.join(SECONDS.toMillis(60));
+ Assertions.assertFalse(workerThread.isAlive(), "worker should have
finished joining");
+ Assertions.assertNull(workerError.get(), "worker should not fail
to start");
+ Assertions.assertTrue(
+ discoverCount.get() > discoveriesBeforeMaster,
+ "worker should have re-polled discovery after the master
became discoverable");
+
+ HazelcastInstanceImpl worker = workerRef.get();
+ Assertions.assertNotNull(worker);
+ Assertions.assertTrue(worker.node.getClusterService().isJoined());
+ Assertions.assertFalse(worker.node.isMaster(), "a worker must
never become master");
+ Assertions.assertTrue(master.node.isMaster());
+ } finally {
+ workerThread.interrupt();
+ try {
+ workerThread.join(SECONDS.toMillis(5));
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ }
+ if (workerRef.get() != null) {
+ workerRef.get().shutdown();
+ }
+ if (master != null) {
+ master.shutdown();
+ }
+ HazelcastInstanceFactory.shutdownAll();
+ }
+ }
+
+ /**
+ * Interrupting a discovery-based worker whose discovery keeps returning
no candidates must stop
+ * the retry loop promptly (rather than spinning until {@code
hazelcast.max.join.seconds}), and
+ * the worker must not join.
+ */
+ @Test
+ void interruptedWorkerWithEmptyCandidatesShouldStopRetrying() throws
Exception {
+ String clusterName =
+
TestUtils.getClusterName("LiteNodeDropOutDiscoveryJoinerTest_interruptEmpty");
+ int masterPort = TestUtils.getAvailablePort(); // never published
+ int workerPort = TestUtils.getAvailablePort();
+
+ CountDownLatch masterReady = new CountDownLatch(1); // never counted
down
+ AtomicInteger discoverCount = new AtomicInteger();
+
+ Config workerConfig =
+ newEmptyDiscoveryConfig(
+ clusterName, workerPort, masterPort, masterReady,
discoverCount);
+ workerConfig.setLiteMember(true);
+ workerConfig.setProperty("hazelcast.max.join.seconds", "60");
+ workerConfig.setProperty("hazelcast.wait.seconds.before.join", "1");
+
+ AtomicReference<HazelcastInstanceImpl> workerRef = new
AtomicReference<>();
+ Thread workerThread =
+ new Thread(
+ () -> {
+ try {
+ workerRef.set(startNode(workerConfig, null));
+ } catch (Throwable ignored) {
+ // The join attempt is expected to fail after
interruption.
+ }
+ },
+ "test-interrupt-empty-candidates");
+ workerThread.start();
+
+ try {
+ // Wait until the worker has passed the initial discovery window
and is polling
+ // discovery from the empty-list refresh loop.
+ long pollDeadline = System.currentTimeMillis() +
SECONDS.toMillis(30);
+ while (discoverCount.get() < 10 && System.currentTimeMillis() <
pollDeadline) {
+ Thread.sleep(100);
+ }
+ Assertions.assertTrue(
+ discoverCount.get() >= 10, "worker should have entered the
refresh loop");
+
+ workerThread.interrupt();
+ workerThread.join(SECONDS.toMillis(5));
+ Assertions.assertFalse(
+ workerThread.isAlive(),
+ "worker should stop retrying promptly after interruption");
+ Assertions.assertNull(workerRef.get(), "worker should not have
joined");
+ } finally {
+ workerThread.interrupt();
+ if (workerRef.get() != null) {
+ workerRef.get().shutdown();
+ }
+ HazelcastInstanceFactory.shutdownAll();
+ }
+ }
+
+ private static HazelcastInstanceImpl startNode(Config config,
CountDownLatch blacklistLatch) {
+ String instanceName =
HazelcastInstanceFactory.createInstanceName(config);
+ HazelcastInstanceProxy proxy =
+ (HazelcastInstanceProxy)
+ HazelcastInstanceFactory.newHazelcastInstance(
+ config,
+ instanceName,
+ new
LiteNodeDropOutDiscoveryJoinerContext(blacklistLatch));
+ return proxy.getOriginal();
+ }
+
+ private static Config newDiscoveryConfig(String clusterName, int
workerPort, int masterPort)
+ throws UnknownHostException {
+ Config config = new Config();
+ config.setClusterName(clusterName);
+ config.getNetworkConfig().setPort(workerPort);
+ config.getNetworkConfig().setPortAutoIncrement(false);
+
config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
+ config.getNetworkConfig().getJoin().getTcpIpConfig().setEnabled(false);
+ DiscoveryConfig discoveryConfig =
config.getNetworkConfig().getJoin().getDiscoveryConfig();
+ discoveryConfig.addDiscoveryStrategyConfig(
+ new DiscoveryStrategyConfig(
+ new StaticDiscoveryStrategyFactory(new Address(HOST,
masterPort))));
+ return config;
+ }
+
+ private static Config newEmptyDiscoveryConfig(
+ String clusterName,
+ int workerPort,
+ int masterPort,
+ CountDownLatch masterReady,
+ AtomicInteger discoverCount)
+ throws UnknownHostException {
+ Config config = new Config();
+ config.setClusterName(clusterName);
+ config.getNetworkConfig().setPort(workerPort);
+ config.getNetworkConfig().setPortAutoIncrement(false);
+
config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
+ config.getNetworkConfig().getJoin().getTcpIpConfig().setEnabled(false);
+ DiscoveryConfig discoveryConfig =
config.getNetworkConfig().getJoin().getDiscoveryConfig();
+ discoveryConfig.addDiscoveryStrategyConfig(
+ new DiscoveryStrategyConfig(
+ new InitiallyEmptyDiscoveryStrategyFactory(
+ new Address(HOST, masterPort), masterReady,
discoverCount)));
+ return config;
+ }
+
+ private static Config newTcpIpConfig(String clusterName, int port) {
+ Config config = new Config();
+ config.setClusterName(clusterName);
+ config.getNetworkConfig().setPort(port);
+ config.getNetworkConfig().setPortAutoIncrement(false);
+
config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
+ TcpIpConfig tcpIpConfig =
config.getNetworkConfig().getJoin().getTcpIpConfig();
+ tcpIpConfig.setEnabled(true);
+ tcpIpConfig.setConnectionTimeoutSeconds(1);
+ return config;
+ }
+
+ /** A discovery joiner that counts down the given latch when an address is
blacklisted. */
+ private static class CountingBlacklistDiscoveryJoiner extends
LiteNodeDropOutDiscoveryJoiner {
+
+ private final CountDownLatch blacklistLatch;
+
+ CountingBlacklistDiscoveryJoiner(Node node, CountDownLatch
blacklistLatch) {
+ super(node);
+ this.blacklistLatch = blacklistLatch;
+ }
+
+ @Override
+ public void blacklist(Address address, boolean permanent) {
+ super.blacklist(address, permanent);
+ if (blacklistLatch != null) {
+ blacklistLatch.countDown();
+ }
+ }
+ }
+
+ /** Node context that selects the discovery joiner for lite members and
TCP/IP otherwise. */
+ private static class LiteNodeDropOutDiscoveryJoinerContext extends
DefaultNodeContext {
+
+ private final CountDownLatch blacklistLatch;
+
+ LiteNodeDropOutDiscoveryJoinerContext(CountDownLatch blacklistLatch) {
+ this.blacklistLatch = blacklistLatch;
+ }
+
+ @Override
+ public Joiner createJoiner(Node node) {
+ return node.isLiteMember()
+ ? new CountingBlacklistDiscoveryJoiner(node,
blacklistLatch)
+ : new LiteNodeDropOutTcpIpJoiner(node);
+ }
+ }
+
+ /** Discovery strategy that always returns the fixed master address. */
+ private static final class StaticDiscoveryStrategy implements
DiscoveryStrategy {
+
+ private final Iterable<DiscoveryNode> nodes;
+
+ private StaticDiscoveryStrategy(Address address) {
+ this.nodes = Collections.singleton(new
SimpleDiscoveryNode(address));
+ }
+
+ @Override
+ public void start() {}
+
+ @Override
+ public Iterable<DiscoveryNode> discoverNodes() {
+ return nodes;
+ }
+
+ @Override
+ public void destroy() {}
+
+ @Override
+ public Map<String, String> discoverLocalMetadata() {
+ return Collections.emptyMap();
+ }
+ }
+
+ private static final class StaticDiscoveryStrategyFactory implements
DiscoveryStrategyFactory {
+
+ private final Address address;
+
+ private StaticDiscoveryStrategyFactory(Address address) {
+ this.address = address;
+ }
+
+ @Override
+ public Class<? extends DiscoveryStrategy> getDiscoveryStrategyType() {
+ return StaticDiscoveryStrategy.class;
+ }
+
+ @Override
+ public DiscoveryStrategy newDiscoveryStrategy(
+ DiscoveryNode discoveryNode, ILogger logger, Map<String,
Comparable> properties) {
+ return new StaticDiscoveryStrategy(address);
+ }
+
+ @Override
+ public Collection<PropertyDefinition> getConfigurationProperties() {
+ return Collections.emptyList();
+ }
+ }
+
+ /**
+ * Discovery strategy that returns no nodes until the given latch is
counted down, then returns
+ * the fixed master address. Models a master that is not published to
discovery until after the
+ * worker has started.
+ */
+ private static final class InitiallyEmptyDiscoveryStrategy implements
DiscoveryStrategy {
+
+ private final Address address;
+ private final CountDownLatch masterReady;
+ private final AtomicInteger discoverCount;
+
+ private InitiallyEmptyDiscoveryStrategy(
+ Address address, CountDownLatch masterReady, AtomicInteger
discoverCount) {
+ this.address = address;
+ this.masterReady = masterReady;
+ this.discoverCount = discoverCount;
+ }
+
+ @Override
+ public void start() {}
+
+ @Override
+ public Iterable<DiscoveryNode> discoverNodes() {
+ discoverCount.incrementAndGet();
+ if (masterReady.getCount() > 0) {
+ return Collections.emptyList();
+ }
+ return Collections.singleton(new SimpleDiscoveryNode(address));
+ }
+
+ @Override
+ public void destroy() {}
+
+ @Override
+ public Map<String, String> discoverLocalMetadata() {
+ return Collections.emptyMap();
+ }
+ }
+
+ private static final class InitiallyEmptyDiscoveryStrategyFactory
+ implements DiscoveryStrategyFactory {
+
+ private final Address address;
+ private final CountDownLatch masterReady;
+ private final AtomicInteger discoverCount;
+
+ private InitiallyEmptyDiscoveryStrategyFactory(
+ Address address, CountDownLatch masterReady, AtomicInteger
discoverCount) {
+ this.address = address;
+ this.masterReady = masterReady;
+ this.discoverCount = discoverCount;
+ }
+
+ @Override
+ public Class<? extends DiscoveryStrategy> getDiscoveryStrategyType() {
+ return InitiallyEmptyDiscoveryStrategy.class;
+ }
+
+ @Override
+ public DiscoveryStrategy newDiscoveryStrategy(
+ DiscoveryNode discoveryNode, ILogger logger, Map<String,
Comparable> properties) {
+ return new InitiallyEmptyDiscoveryStrategy(address, masterReady,
discoverCount);
+ }
+
+ @Override
+ public Collection<PropertyDefinition> getConfigurationProperties() {
+ return Collections.emptyList();
+ }
+ }
+}
diff --git
a/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutTcpIpJoinerTest.java
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutTcpIpJoinerTest.java
new file mode 100644
index 0000000000..3698c34208
--- /dev/null
+++
b/seatunnel-engine/seatunnel-engine-server/src/test/java/org/apache/seatunnel/engine/server/joiner/LiteNodeDropOutTcpIpJoinerTest.java
@@ -0,0 +1,382 @@
+/*
+ * 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.seatunnel.engine.server.joiner;
+
+import org.apache.seatunnel.engine.server.TestUtils;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import com.hazelcast.cluster.Address;
+import com.hazelcast.config.Config;
+import com.hazelcast.config.TcpIpConfig;
+import com.hazelcast.instance.impl.DefaultNodeContext;
+import com.hazelcast.instance.impl.HazelcastInstanceFactory;
+import com.hazelcast.instance.impl.HazelcastInstanceImpl;
+import com.hazelcast.instance.impl.HazelcastInstanceProxy;
+import com.hazelcast.instance.impl.Node;
+import com.hazelcast.internal.cluster.Joiner;
+
+import java.util.Arrays;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static java.util.concurrent.TimeUnit.SECONDS;
+
+/**
+ * Verifies the lite-member (worker) join behavior of {@link
LiteNodeDropOutTcpIpJoiner}.
+ *
+ * <p>A worker is a Hazelcast lite member and therefore can never assume the
master role. The
+ * regression being guarded here is: when a worker starts before the master,
connecting to the
+ * candidate master addresses fails and they are temporarily blacklisted.
Without the retry fix the
+ * worker would never retry those addresses and would deadlock until {@code
+ * hazelcast.max.join.seconds} elapses.
+ */
+class LiteNodeDropOutTcpIpJoinerTest {
+
+ private static final String HOST = "127.0.0.1";
+
+ private static Config newConfig(String clusterName, int port, String...
members) {
+ Config config = new Config();
+ config.setClusterName(clusterName);
+ config.getNetworkConfig().setPort(port);
+ config.getNetworkConfig().setPortAutoIncrement(false);
+
config.getNetworkConfig().getJoin().getMulticastConfig().setEnabled(false);
+ TcpIpConfig tcpIpConfig =
config.getNetworkConfig().getJoin().getTcpIpConfig();
+ tcpIpConfig.setEnabled(true);
+ // Keep the connection timeout small so that a failed connection
attempt against a
+ // not-yet-started master is detected and blacklisted quickly, making
the test
+ // deterministic.
+ tcpIpConfig.setConnectionTimeoutSeconds(1);
+ tcpIpConfig.setMembers(Arrays.asList(members));
+ return config;
+ }
+
+ private static HazelcastInstanceImpl startNode(Config config,
CountDownLatch blacklistLatch) {
+ String instanceName =
HazelcastInstanceFactory.createInstanceName(config);
+ HazelcastInstanceProxy proxy =
+ (HazelcastInstanceProxy)
+ HazelcastInstanceFactory.newHazelcastInstance(
+ config,
+ instanceName,
+ new
LiteNodeDropOutJoinerContext(blacklistLatch));
+ return proxy.getOriginal();
+ }
+
+ /**
+ * A worker (lite member) starts before the master. The worker initially
fails to connect to the
+ * not-yet-started master and temporarily blacklists its address, then
retries. Once the master
+ * is started (before the join timeout), the worker must join the cluster
and must never become
+ * master itself.
+ *
+ * <p>The test starts the master only after the worker has been observed
to blacklist the master
+ * address (signaled via a latch), guaranteeing the temporary-blacklist
retry path is exercised
+ * rather than a plain first-attempt join.
+ */
+ @Test
+ void workerStartedBeforeMasterShouldEventuallyJoin() throws Exception {
+ String clusterName =
TestUtils.getClusterName("LiteNodeDropOutTcpIpJoinerTest_workerJoin");
+ int masterPort = TestUtils.getAvailablePort();
+ int workerPort = TestUtils.getAvailablePort();
+
+ Config workerConfig = newConfig(clusterName, workerPort, HOST + ":" +
masterPort);
+ workerConfig.setLiteMember(true);
+ // Generous join window; the worker is expected to join well before
this expires.
+ workerConfig.setProperty("hazelcast.max.join.seconds", "60");
+
+ // Fires once the worker has actually blacklisted the not-yet-started
master address.
+ CountDownLatch blacklistLatch = new CountDownLatch(1);
+
+ AtomicReference<HazelcastInstanceImpl> workerRef = new
AtomicReference<>();
+ AtomicReference<Throwable> workerError = new AtomicReference<>();
+ Thread workerThread =
+ new Thread(
+ () -> {
+ try {
+ workerRef.set(startNode(workerConfig,
blacklistLatch));
+ } catch (Throwable t) {
+ workerError.set(t);
+ }
+ },
+ "test-worker-started-before-master");
+ workerThread.start();
+
+ HazelcastInstanceImpl master = null;
+ try {
+ // Wait until the worker has provably blacklisted the
not-yet-started master address.
+ Assertions.assertTrue(
+ blacklistLatch.await(30, SECONDS),
+ "worker should have blacklisted the not-yet-started master
address");
+
+ Config masterConfig = newConfig(clusterName, masterPort);
+ master = startNode(masterConfig, null);
+
+ workerThread.join(SECONDS.toMillis(60));
+ Assertions.assertFalse(workerThread.isAlive(), "worker should have
finished joining");
+ Assertions.assertNull(workerError.get(), "worker should not fail
to start");
+
+ HazelcastInstanceImpl worker = workerRef.get();
+ Assertions.assertNotNull(worker);
+ Assertions.assertTrue(worker.node.getClusterService().isJoined());
+ Assertions.assertFalse(worker.node.isMaster(), "a worker must
never become master");
+ Assertions.assertTrue(master.node.isMaster());
+ } finally {
+ workerThread.interrupt();
+ try {
+ workerThread.join(SECONDS.toMillis(5));
+ } catch (InterruptedException ignored) {
+ Thread.currentThread().interrupt();
+ }
+ if (workerRef.get() != null) {
+ workerRef.get().shutdown();
+ }
+ if (master != null) {
+ master.shutdown();
+ }
+ // Force-stop any still-starting worker node (e.g. when a timeout
assertion failed
+ // before workerRef was set) so it cannot leak into later tests in
the same JVM.
+ HazelcastInstanceFactory.shutdownAll();
+ }
+ }
+
+ /**
+ * A worker (lite member) with no master available must time out and must
never assume the
+ * master role. If the worker incorrectly self-elected, {@code startNode}
would return
+ * successfully instead of throwing.
+ */
+ @Test
+ void workerWithoutMasterShouldTimeoutAndNeverBecomeMaster() {
+ String clusterName =
+
TestUtils.getClusterName("LiteNodeDropOutTcpIpJoinerTest_workerTimeout");
+ int masterPort = TestUtils.getAvailablePort(); // never started
+ int workerPort = TestUtils.getAvailablePort();
+
+ Config workerConfig = newConfig(clusterName, workerPort, HOST + ":" +
masterPort);
+ workerConfig.setLiteMember(true);
+ workerConfig.setProperty("hazelcast.max.join.seconds", "5");
+
+ Assertions.assertThrows(IllegalStateException.class, () ->
startNode(workerConfig, null));
+ }
+
+ /**
+ * The non-lite member path must remain unchanged. A non-lite member whose
candidate addresses
+ * are all blacklisted (for example, a dead target address) must still
assume the master role
+ * instead of entering the lite-member retry loop.
+ */
+ @Test
+ void nonLiteMemberShouldAssumeMasterWhenAllAddressesBlacklisted() {
+ String clusterName =
+
TestUtils.getClusterName("LiteNodeDropOutTcpIpJoinerTest_nonLiteSelfElect");
+ int deadTargetPort = TestUtils.getAvailablePort(); // never started
+ int memberPort = TestUtils.getAvailablePort();
+
+ Config config = newConfig(clusterName, memberPort, HOST + ":" +
deadTargetPort);
+ config.setProperty("hazelcast.max.join.seconds", "30");
+
+ HazelcastInstanceImpl instance = startNode(config, null);
+ try {
+
Assertions.assertTrue(instance.node.getClusterService().isJoined());
+ Assertions.assertTrue(instance.node.isMaster());
+ } finally {
+ instance.shutdown();
+ }
+ }
+
+ /** A reachable master must stay excluded when its address is permanently
blacklisted. */
+ @Test
+ void retryMustNotClearAPermanentBlacklistEvenWhenTheMasterIsReachable()
throws Exception {
+ String clusterName =
TestUtils.getClusterName("LiteNodeDropOutTcpIpJoinerTest_permanent");
+ int masterPort = TestUtils.getAvailablePort();
+ int workerPort = TestUtils.getAvailablePort();
+ Address masterAddress = new Address(HOST, masterPort);
+ HazelcastInstanceImpl master = startNode(newConfig(clusterName,
masterPort), null);
+ AtomicReference<HazelcastInstanceImpl> unexpectedWorker = new
AtomicReference<>();
+ try {
+ Config workerConfig = newConfig(clusterName, workerPort, HOST +
":" + masterPort);
+ workerConfig.setLiteMember(true);
+ workerConfig.setProperty("hazelcast.max.join.seconds", "15");
+ Assertions.assertThrows(
+ IllegalStateException.class,
+ () -> {
+ String instanceName =
+
HazelcastInstanceFactory.createInstanceName(workerConfig);
+ HazelcastInstanceProxy proxy =
+ (HazelcastInstanceProxy)
+
HazelcastInstanceFactory.newHazelcastInstance(
+ workerConfig,
+ instanceName,
+ new DefaultNodeContext() {
+ @Override
+ public Joiner
createJoiner(Node node) {
+ return new
LiteNodeDropOutTcpIpJoiner(
+ node) {
+ @Override
+ public void
doJoin() {
+ //
AbstractJoiner.join clears
+ // old entries
before doJoin, so
+ // inject the
permanent exclusion
+ // here.
+
blacklist(masterAddress, true);
+ super.doJoin();
+ }
+ };
+ }
+ });
+ unexpectedWorker.set(proxy.getOriginal());
+ });
+ Assertions.assertTrue(master.node.getClusterService().isJoined());
+ Assertions.assertTrue(master.node.isMaster());
+ Assertions.assertEquals(1,
master.node.getClusterService().getMembers().size());
+ } finally {
+ if (unexpectedWorker.get() != null) {
+ unexpectedWorker.get().shutdown();
+ }
+ master.shutdown();
+ }
+ }
+
+ /**
+ * Interrupting a worker whose candidates are all permanently blacklisted
must stop the retry
+ * loop promptly (rather than spinning until {@code
hazelcast.max.join.seconds}), and the worker
+ * must not join.
+ */
+ @Test
+ void interruptedWorkerWithPermanentBlacklistShouldStopRetrying() throws
Exception {
+ String clusterName =
+
TestUtils.getClusterName("LiteNodeDropOutTcpIpJoinerTest_interruptPermanent");
+ int masterPort = TestUtils.getAvailablePort(); // never started
+ int workerPort = TestUtils.getAvailablePort();
+ Address masterAddress = new Address(HOST, masterPort);
+
+ // Counts down once the retry loop reaches the unblacklist step, so
the interrupt lands
+ // deterministically while the worker is in the
all-permanent-blacklist retry state.
+ CountDownLatch retryLatch = new CountDownLatch(1);
+
+ Config workerConfig = newConfig(clusterName, workerPort, HOST + ":" +
masterPort);
+ workerConfig.setLiteMember(true);
+ workerConfig.setProperty("hazelcast.max.join.seconds", "60");
+
+ AtomicReference<HazelcastInstanceImpl> workerRef = new
AtomicReference<>();
+ Thread workerThread =
+ new Thread(
+ () -> {
+ try {
+ String instanceName =
+
HazelcastInstanceFactory.createInstanceName(workerConfig);
+ HazelcastInstanceProxy proxy =
+ (HazelcastInstanceProxy)
+
HazelcastInstanceFactory.newHazelcastInstance(
+ workerConfig,
+ instanceName,
+ new
DefaultNodeContext() {
+ @Override
+ public Joiner
createJoiner(Node node) {
+ return new
LiteNodeDropOutTcpIpJoiner(
+ node) {
+ @Override
+ public
void doJoin() {
+ //
AbstractJoiner.join
+ //
clears old entries
+ //
before doJoin, so inject
+ // the
permanent exclusion
+ //
here.
+
blacklist(
+
masterAddress,
+
true);
+
super.doJoin();
+ }
+
+ @Override
+ public
boolean unblacklist(
+
Address address) {
+
boolean result =
+
super.unblacklist(
+
address);
+
retryLatch.countDown();
+ return
result;
+ }
+ };
+ }
+ });
+ workerRef.set(proxy.getOriginal());
+ } catch (Throwable ignored) {
+ // The join attempt is expected to fail after
interruption.
+ }
+ },
+ "test-interrupt-permanent-blacklist");
+ workerThread.start();
+
+ try {
+ Assertions.assertTrue(
+ retryLatch.await(30, SECONDS),
+ "worker should have entered the permanent-blacklist retry
state");
+
+ workerThread.interrupt();
+ workerThread.join(SECONDS.toMillis(5));
+ Assertions.assertFalse(
+ workerThread.isAlive(),
+ "worker should stop retrying promptly after interruption");
+ Assertions.assertNull(workerRef.get(), "worker should not have
joined");
+ } finally {
+ workerThread.interrupt();
+ if (workerRef.get() != null) {
+ workerRef.get().shutdown();
+ }
+ HazelcastInstanceFactory.shutdownAll();
+ }
+ }
+
+ /**
+ * A joiner that counts down the given latch whenever an address is
blacklisted, so tests can
+ * deterministically wait for the worker to reach the temporary-blacklist
state.
+ */
+ private static class CountingBlacklistJoiner extends
LiteNodeDropOutTcpIpJoiner {
+
+ private final CountDownLatch blacklistLatch;
+
+ CountingBlacklistJoiner(Node node, CountDownLatch blacklistLatch) {
+ super(node);
+ this.blacklistLatch = blacklistLatch;
+ }
+
+ @Override
+ public void blacklist(Address address, boolean permanent) {
+ super.blacklist(address, permanent);
+ if (blacklistLatch != null) {
+ blacklistLatch.countDown();
+ }
+ }
+ }
+
+ /**
+ * Minimal {@link com.hazelcast.instance.impl.NodeContext} that wires in
the joiner under test.
+ */
+ private static class LiteNodeDropOutJoinerContext extends
DefaultNodeContext {
+
+ private final CountDownLatch blacklistLatch;
+
+ LiteNodeDropOutJoinerContext(CountDownLatch blacklistLatch) {
+ this.blacklistLatch = blacklistLatch;
+ }
+
+ @Override
+ public Joiner createJoiner(Node node) {
+ return new CountingBlacklistJoiner(node, blacklistLatch);
+ }
+ }
+}
