This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12755-c389c24b0ec24c55dafbf2d0e9d98189474d3e2f in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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); + } + } +}
