This is an automated email from the ASF dual-hosted git repository.
petrov-mg pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new 98a0a8c5470 IGNITE-28883 Stabilized Ignite Features related tests
(#13360)
98a0a8c5470 is described below
commit 98a0a8c547029af1073affa3633e329c36f70b9f
Author: Mikhail Petrov <[email protected]>
AuthorDate: Tue Jul 14 09:21:19 2026 +0300
IGNITE-28883 Stabilized Ignite Features related tests (#13360)
---
.../rollingupgrade/RollingUpgradeClusterData.java | 6 -
.../rollingupgrade/RollingUpgradeProcessor.java | 32 +-
.../ignite/spi/discovery/tcp/ServerImpl.java | 32 +-
.../ignite/spi/discovery/tcp/TcpDiscoverySpi.java | 15 +-
.../rollingupgrade/AbstractRollingUpgradeTest.java | 5 +-
.../CoreVersionRollingUpgradeTest.java | 104 +++---
.../NodeSecurityContextPropagationTest.java | 348 +--------------------
.../discovery/tcp/MultiDataCenterSplitTest.java | 11 +-
.../tcp/TcpDiscoveryNetworkIssuesTest.java | 12 +-
.../TcpDiscoveryPendingMessageDeliveryTest.java | 31 +-
.../discovery/tcp/TestBlockingTcpDiscoverySpi.java | 122 ++++++++
11 files changed, 241 insertions(+), 477 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeClusterData.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeClusterData.java
index b6779d47560..f28dc98558a 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeClusterData.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeClusterData.java
@@ -35,10 +35,6 @@ public class RollingUpgradeClusterData implements Message {
/** */
@Order(2)
- boolean isNodeFenceActive;
-
- /** */
- @Order(3)
Collection<IgniteComponentFeatureSet> activeFeatures;
/** */
@@ -50,12 +46,10 @@ public class RollingUpgradeClusterData implements Message {
public RollingUpgradeClusterData(
boolean isVersionUpgradeEnabled,
UUID curFinalizeProcId,
- boolean isNodeFenceActive,
Collection<IgniteComponentFeatureSet> activeFeatures
) {
this.isVersionUpgradeEnabled = isVersionUpgradeEnabled;
this.curFinalizeProcId = curFinalizeProcId;
- this.isNodeFenceActive = isNodeFenceActive;
this.activeFeatures = activeFeatures;
}
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java
index 9268f89f550..ab2f9cda1d9 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/RollingUpgradeProcessor.java
@@ -88,9 +88,6 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
/** */
private final Set<ClusterNode> joiningNodes = new HashSet<>();
- /** */
- private volatile boolean isNodeFenceActive;
-
/** */
private volatile boolean isVerUpgradeEnabled;
@@ -233,7 +230,6 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
isVerUpgradeEnabled = gridData.isVersionUpgradeEnabled;
curFinalizeProcId = gridData.curFinalizeProcId;
- isNodeFenceActive = gridData.isNodeFenceActive;
featureMgr.onGridDataReceived(new
IgniteNodeFeatureSet(gridData.activeFeatures));
}
@@ -241,7 +237,7 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
/** {@inheritDoc} */
@Override public @Nullable IgniteNodeValidationResult
validateNode(ClusterNode joiningNode) {
synchronized (topGuard) {
- if (isNodeFenceActive) {
+ if (curFinalizeProcId != null) {
return new IgniteNodeValidationResult(
joiningNode.id(),
"Node joins are not allowed during cluster version
finalization. Retry the node join procedure after" +
@@ -408,7 +404,6 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
return new RollingUpgradeClusterData(
isVerUpgradeEnabled,
curFinalizeProcId,
- isNodeFenceActive,
featureMgr.activeFeatures().values()
);
}
@@ -543,9 +538,6 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
if (err != null)
U.error(log, "Cluster version finalization process failed
[procId=" + reqId + ']', err);
- if (reqId.equals(curFinalizeProcId))
- curFinalizeProcId = null;
-
super.finishProcess(reqId, err);
}
@@ -554,7 +546,6 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
U.warn(log, "Cluster version finalization process has been aborted
[procId=" + curFinalizeProcId + ", reason=" + reason + ']');
curFinalizeProcId = null;
- isNodeFenceActive = false;
super.abort(reason);
}
@@ -566,16 +557,16 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
/** */
private IgniteInternalFuture<Message> executePreparePhase(UUID reqId,
Message req) {
- if (curFinalizeProcId != null) {
- U.error(log, "Failed to handle cluster version finalization
request. Another process is " +
- "already in progress [curProcId=" + reqId + ",
activeProcId=" + curFinalizeProcId + ']');
+ synchronized (topGuard) {
+ if (curFinalizeProcId != null) {
+ U.error(log, "Failed to handle cluster version
finalization request. Another process is " +
+ "already in progress [curProcId=" + reqId + ",
activeProcId=" + curFinalizeProcId + ']');
- return new GridFinishedFuture<>(new IgniteException("Cluster
version finalization process is already in progress"));
- }
+ return new GridFinishedFuture<>(new
IgniteException("Cluster version finalization process is already in progress"));
+ }
- curFinalizeProcId = reqId;
+ curFinalizeProcId = reqId;
- synchronized (topGuard) {
if (!isReadyForVersionFinalization())
return new GridFinishedFuture<>(new IgniteException(
"Cluster version finalization failed. The cluster contains
nodes running" +
@@ -584,8 +575,6 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
" [clusterFeatures=" +
clusterFeatures().entrySet().stream().collect(
Collectors.toMap(e -> e.getKey().id(),
Map.Entry::getValue)) + ']'));
- isNodeFenceActive = true;
-
return new GridFinishedFuture<>();
}
}
@@ -594,7 +583,7 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
private void finishPreparePhase(UUID reqId, Map<UUID, Message>
responses, Map<UUID, Throwable> errors) {
if (!F.isEmpty(errors)) {
if (reqId.equals(curFinalizeProcId))
- isNodeFenceActive = false;
+ curFinalizeProcId = null;
finishProcess(reqId, firstError(errors));
}
@@ -618,8 +607,7 @@ public class RollingUpgradeProcessor extends
GridProcessorAdapter implements Dis
featureMgr.activateLocalVersionFeatures();
isVerUpgradeEnabled = false;
-
- isNodeFenceActive = false;
+ curFinalizeProcId = null;
return new GridFinishedFuture<>();
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
index 9af1a4200d6..c95bba54b6c 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/ServerImpl.java
@@ -2949,11 +2949,14 @@ class ServerImpl extends TcpDiscoveryImpl {
/** Thread local variable indicates that discovery manager was
notified after message processing. */
private final ThreadLocal<Boolean> notifiedDiscovery =
ThreadLocal.withInitial(() -> false);
- /**
- * @param log Logger.
- */
- protected RingMessageWorker(IgniteLogger log) {
- super("tcp-disco-msg-worker-[]", log, 10, getWorkerRegistry(spi));
+ /** */
+ RingMessageWorker(IgniteLogger log) {
+ this(log, new LinkedBlockingDeque<>());
+ }
+
+ /** */
+ protected RingMessageWorker(IgniteLogger log,
BlockingDeque<TcpDiscoveryAbstractMessage> queue) {
+ super("tcp-disco-msg-worker-[]", log, 10, getWorkerRegistry(spi),
queue);
setBeforeEachPollAction(() -> {
updateHeartbeat();
@@ -7775,9 +7778,15 @@ class ServerImpl extends TcpDiscoveryImpl {
* @param log Logger.
*/
private ClientMessageWorker(Socket sock, UUID clientNodeId,
IgniteLogger log) {
- super("tcp-disco-client-message-worker-[" + U.id8(clientNodeId)
- + ' ' + sock.getInetAddress().getHostAddress()
- + ":" + sock.getPort() + ']', log,
Math.max(spi.metricsUpdateFreq, 10), null);
+ super(
+ "tcp-disco-client-message-worker-[" + U.id8(clientNodeId) +
+ ' ' + sock.getInetAddress().getHostAddress() +
+ ":" + sock.getPort() + ']',
+ log,
+ Math.max(spi.metricsUpdateFreq, 10),
+ null,
+ new LinkedBlockingDeque<>()
+ );
this.sock = sock;
this.clientNodeId = clientNodeId;
@@ -8104,7 +8113,7 @@ class ServerImpl extends TcpDiscoveryImpl {
*/
private abstract class MessageWorker<T> extends GridWorker {
/** Message queue. */
- protected final BlockingDeque<T> queue = new LinkedBlockingDeque<>();
+ protected final BlockingDeque<T> queue;
/** Polling timeout. */
private final long pollingTimeout;
@@ -8117,15 +8126,18 @@ class ServerImpl extends TcpDiscoveryImpl {
* @param log Logger.
* @param pollingTimeout Messages polling timeout.
* @param lsnr Listener for life-cycle events.
+ * @param queue The queue used for holding messages and handing off to
worker thread.
*/
protected MessageWorker(
String name,
IgniteLogger log,
long pollingTimeout,
- @Nullable GridWorkerListener lsnr
+ @Nullable GridWorkerListener lsnr,
+ BlockingDeque<T> queue
) {
super(spi.ignite().name(), name, log, lsnr);
+ this.queue = queue;
this.pollingTimeout = pollingTimeout;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
index eb868103ad1..5594e77395b 100644
---
a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
+++
b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java
@@ -445,7 +445,7 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
protected IgniteLogger log;
/** */
- protected TcpDiscoveryImpl impl;
+ TcpDiscoveryImpl impl;
/** */
private boolean clientReconnectDisabled;
@@ -2118,10 +2118,8 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
impl.spiStart(igniteInstanceName);
}
- /**
- *
- */
- protected void initializeImpl() {
+ /** */
+ private void initializeImpl() {
if (impl != null)
return;
@@ -2147,7 +2145,7 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
if (sockTimeout == 0)
sockTimeout = DFLT_SOCK_TIMEOUT;
- impl = new ServerImpl(this, DFLT_UTLITY_POOL_SIZE,
DFLT_RMT_DC_PING_POOL_SIZE);
+ impl = createServerTcpDiscoveryImplementation();
}
metricsUpdateFreq = ignite.configuration().getMetricsUpdateFrequency();
@@ -2230,6 +2228,11 @@ public class TcpDiscoverySpi extends IgniteSpiAdapter
implements IgniteDiscovery
cfgNodeId = ignite.configuration().getNodeId();
}
+ /** */
+ TcpDiscoveryImpl createServerTcpDiscoveryImplementation() {
+ return new ServerImpl(this, DFLT_UTLITY_POOL_SIZE,
DFLT_RMT_DC_PING_POOL_SIZE);
+ }
+
/** {@inheritDoc} */
@Override public void spiStop() throws IgniteSpiException {
if (ctxInitLatch.getCount() > 0)
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractRollingUpgradeTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractRollingUpgradeTest.java
index f34a9ea6347..98718dfb635 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractRollingUpgradeTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/AbstractRollingUpgradeTest.java
@@ -64,6 +64,7 @@ import org.apache.ignite.plugin.PluginContext;
import org.apache.ignite.spi.IgniteNodeValidationResult;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.TestBlockingTcpDiscoverySpi;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.jspecify.annotations.Nullable;
@@ -207,14 +208,12 @@ public abstract class AbstractRollingUpgradeTest extends
GridCommonAbstractTest
}
});
- TcpDiscoverySpi discoSpi = new TcpDiscoverySpi() {
+ TcpDiscoverySpi discoSpi = new TestBlockingTcpDiscoverySpi(cfg) {
@Override public void setNodeAttributes(Map<String, Object> attrs,
IgniteProductVersion ignored) {
super.setNodeAttributes(attrs,
IgniteProductVersion.fromString(testVersions.coreVersion()));
}
};
-
discoSpi.setIpFinder(((TcpDiscoverySpi)cfg.getDiscoverySpi()).getIpFinder());
-
cfg.setDiscoverySpi(discoSpi);
return cfg;
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/CoreVersionRollingUpgradeTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/CoreVersionRollingUpgradeTest.java
index 82eb8d85990..e852649829a 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/CoreVersionRollingUpgradeTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/rollingupgrade/CoreVersionRollingUpgradeTest.java
@@ -43,8 +43,7 @@ import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.ignite.events.EventType.EVT_CLIENT_NODE_RECONNECTED;
import static org.apache.ignite.internal.TestRecordingCommunicationSpi.spi;
import static
org.apache.ignite.internal.processors.rollingupgrade.feature.TestIgniteReleaseFeatures_2_19_2.VER_2_19_2_ID_1_FEATURE;
-import static
org.apache.ignite.internal.processors.security.NodeSecurityContextPropagationTest.discoveryRingMessageWorkerQueue;
-import static
org.apache.ignite.internal.processors.security.NodeSecurityContextPropagationTest.wrapRingMessageWorkerQueue;
+import static
org.apache.ignite.spi.discovery.tcp.TestBlockingTcpDiscoverySpi.blockingDiscovery;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
/** */
@@ -416,9 +415,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
/** */
@Test
public void testValidatedJoiningNodesAccountedDuringFinalization() throws
Exception {
- wrapRingMessageWorkerQueue(startGrid(0, TEST_DEFAULT_VER));
- startGrid(1, TEST_DEFAULT_VER);
- startClientGrid(2, TEST_DEFAULT_VER);
+ startCluster();
ru(1).enableVersionUpgrade();
@@ -430,7 +427,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
assertTrue(TestRollingUpgradeProcessor.nodeJoinValidationCompletedLatch.await(getTestTimeout(),
MILLISECONDS));
- discoveryRingMessageWorkerQueue(grid(0)).block();
+ blockingDiscovery(grid(0)).block();
IgniteInternalFuture<Object> finalizeFut =
GridTestUtils.runAsync(() -> finalizeClusterVersion(1, TEST_DEFAULT_VER));
@@ -438,7 +435,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
TestRollingUpgradeProcessor.nodeJoinUnblockedLatch.countDown();
- discoveryRingMessageWorkerQueue(grid(0)).unblock();
+ blockingDiscovery(grid(0)).unblock();
GridTestUtils.assertThrowsAnyCause(
log,
@@ -507,7 +504,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
/** */
@Test
public void testConcurrentFinalizationErrorPreserveNodeFence() throws
Exception {
- wrapRingMessageWorkerQueue(startGrid(0, TEST_DEFAULT_VER));
+ startGrid(0, TEST_DEFAULT_VER);
startGrid(1, TEST_DEFAULT_VER);
startGrid(2, TEST_DEFAULT_VER);
@@ -515,7 +512,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
ru(1).enableVersionUpgrade();
- discoveryRingMessageWorkerQueue(grid(0)).block();
+ blockingDiscovery(grid(0)).block();
try {
IgniteInternalFuture<Object> firstFut = GridTestUtils.runAsync(()
-> finalizeClusterVersion(1, TEST_DEFAULT_VER));
@@ -535,7 +532,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
return singleNodeMsg.processId().equals(startedFinalizeProcId);
});
- discoveryRingMessageWorkerQueue(grid(0)).unblock();
+ blockingDiscovery(grid(0)).unblock();
spi(grid(2)).waitForBlocked();
@@ -553,7 +550,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
firstFut.get(getTestTimeout(), MILLISECONDS);
}
finally {
- discoveryRingMessageWorkerQueue(grid(0)).unblock();
+ blockingDiscovery(grid(0)).unblock();
spi(grid(2)).stopBlock();
}
}
@@ -591,12 +588,12 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
@Test
public void testClientNodeClearsActiveFinalizationProcessOnDisconnect()
throws Exception {
startCluster();
- wrapRingMessageWorkerQueue(startClientGrid(3, TEST_DEFAULT_VER));
+ startClientGrid(3, TEST_DEFAULT_VER);
- CountDownLatch latch = new CountDownLatch(1);
+ CountDownLatch cliReconnectedLatch = new CountDownLatch(1);
grid(3).events().localListen(evt -> {
- latch.countDown();
+ cliReconnectedLatch.countDown();
return true;
}, EVT_CLIENT_NODE_RECONNECTED);
@@ -609,16 +606,31 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
spi(grid(3)).waitForBlocked();
- discoveryRingMessageWorkerQueue(grid(3)).block();
+ AtomicReference<TcpDiscoveryJoinRequestMessage> cliJoinReq = new
AtomicReference<>();
+ CountDownLatch cliJoinReqReceivedLatch = new CountDownLatch(1);
+
+ blockingDiscovery(grid(0)).messageFilter(m -> {
+ if ((m instanceof TcpDiscoveryJoinRequestMessage joinReq) &&
joinReq.node().isClient()) {
+ cliJoinReq.set(joinReq);
+ cliJoinReqReceivedLatch.countDown();
+
+ return true;
+ }
+
+ return false;
+ });
grid(0).context().discovery().failNode(grid(3).context().localNodeId(), "test");
finalizeFut.get(getTestTimeout(), MILLISECONDS);
- spi(grid(3)).stopBlock();
- discoveryRingMessageWorkerQueue(grid(3)).unblock();
+ assertTrue(cliJoinReqReceivedLatch.await(getTestTimeout(),
MILLISECONDS));
- assertTrue(latch.await(getTestTimeout(), MILLISECONDS));
+ blockingDiscovery(grid(0)).messageQueue().addLast(cliJoinReq.get());
+
+ assertTrue(cliReconnectedLatch.await(getTestTimeout(), MILLISECONDS));
+
+ spi(grid(3)).stopBlock();
ru(1).enableVersionUpgrade();
@@ -666,13 +678,11 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
/** */
@Test
public void testConcurrentAbortAndPrepareFinalization() throws Exception {
- wrapRingMessageWorkerQueue(startGrid(0, TEST_DEFAULT_VER));
- startGrid(1, TEST_DEFAULT_VER);
- startClientGrid(2, TEST_DEFAULT_VER);
+ startCluster();
ru(2).enableVersionUpgrade();
- discoveryRingMessageWorkerQueue(grid(0)).block();
+ blockingDiscovery(grid(0)).block();
IgniteInternalFuture<Object> abortFut = GridTestUtils.runAsync(() ->
ru(0).abortClusterVersionFinalization());
@@ -682,7 +692,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
waitForBlockedDiscoveryMessages(grid(0), 2, InitMessage.class);
- discoveryRingMessageWorkerQueue(grid(0)).unblock();
+ blockingDiscovery(grid(0)).unblock();
abortFut.get(getTestTimeout(), MILLISECONDS);
finalizeFut.get(getTestTimeout(), MILLISECONDS);
@@ -691,9 +701,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
/** */
@Test
public void testAbortAfterFinalizationPrepare() throws Exception {
- wrapRingMessageWorkerQueue(startGrid(0, TEST_DEFAULT_VER));
- startGrid(1, TEST_DEFAULT_VER);
- startClientGrid(2, TEST_DEFAULT_VER);
+ startCluster();
ru(2).enableVersionUpgrade();
@@ -703,13 +711,13 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
spi(grid(1)).waitForBlocked();
- discoveryRingMessageWorkerQueue(grid(0)).block();
+ blockingDiscovery(grid(0)).block();
IgniteInternalFuture<Object> abortFut = GridTestUtils.runAsync(() ->
ru(2).abortClusterVersionFinalization());
waitForBlockedDiscoveryMessages(grid(0), 1, InitMessage.class);
- discoveryRingMessageWorkerQueue(grid(0)).unblock();
+ blockingDiscovery(grid(0)).unblock();
spi(grid(1)).stopBlock();
abortFut.get(getTestTimeout(), MILLISECONDS);
@@ -730,9 +738,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
/** */
@Test
public void testAbortBeforeFinalizationComplete() throws Exception {
- wrapRingMessageWorkerQueue(startGrid(0, TEST_DEFAULT_VER));
- startGrid(1, TEST_DEFAULT_VER);
- startClientGrid(2, TEST_DEFAULT_VER);
+ startCluster();
ru(2).enableVersionUpgrade();
@@ -747,7 +753,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
CountDownLatch finalizeCompleteStartedLatch = new CountDownLatch(1);
AtomicReference<TcpDiscoveryAbstractMessage> finalizeCompleteStartMsg
= new AtomicReference<>();
- discoveryRingMessageWorkerQueue(grid(0)).startMessageIntercepting(m ->
{
+ blockingDiscovery(grid(0)).messageFilter(m -> {
if ((m instanceof TcpDiscoveryCustomEventMessage customMsg)
&& (customMsg.message() instanceof InitMessage<?> initMsg)
&& initMsg.processId().equals(startedFinalizeProcId)
@@ -755,25 +761,25 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
finalizeCompleteStartMsg.set(m);
finalizeCompleteStartedLatch.countDown();
- return false;
+ return true;
}
- return true;
+ return false;
});
spi(grid(1)).stopBlock();
assertTrue(finalizeCompleteStartedLatch.await(getTestTimeout(),
MILLISECONDS));
- discoveryRingMessageWorkerQueue(grid(0)).block();
+ blockingDiscovery(grid(0)).block();
IgniteInternalFuture<Object> abortFut = GridTestUtils.runAsync(() ->
ru(2).abortClusterVersionFinalization());
waitForBlockedDiscoveryMessages(grid(0), 1, InitMessage.class);
-
discoveryRingMessageWorkerQueue(grid(0)).addLast(finalizeCompleteStartMsg.get());
+
blockingDiscovery(grid(0)).messageQueue().addLast(finalizeCompleteStartMsg.get());
- discoveryRingMessageWorkerQueue(grid(0)).unblock();
+ blockingDiscovery(grid(0)).unblock();
abortFut.get(getTestTimeout(), MILLISECONDS);
@@ -793,7 +799,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
/** */
@Test
public void testAbortedAfterFinalizationComplete() throws Exception {
- wrapRingMessageWorkerQueue(startGrid(0, TEST_DEFAULT_VER));
+ startGrid(0, TEST_DEFAULT_VER);
startGrid(1, TEST_DEFAULT_VER);
startGrid(2, TEST_DEFAULT_VER);
startClientGrid(3, TEST_DEFAULT_VER);
@@ -806,7 +812,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
spi(grid(1)).waitForBlocked();
- discoveryRingMessageWorkerQueue(grid(0)).block();
+ blockingDiscovery(grid(0)).block();
spi(grid(1)).stopBlock();
@@ -814,7 +820,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
spi(grid(1)).blockMessages((node, msg) -> msg instanceof
SingleNodeMessage);
- discoveryRingMessageWorkerQueue(grid(0)).unblock();
+ blockingDiscovery(grid(0)).unblock();
spi(grid(1)).waitForBlocked();
@@ -856,9 +862,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
/** */
@Test
public void testConcurrentNodeJoinAndFinalization() throws Exception {
- wrapRingMessageWorkerQueue(startGrid(0, TEST_DEFAULT_VER));
- startGrid(1, TEST_DEFAULT_VER);
- startClientGrid(2, TEST_DEFAULT_VER);
+ startCluster();
ru(1).enableVersionUpgrade();
@@ -870,7 +874,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
assertTrue(TestRollingUpgradeProcessor.nodeJoinValidationCompletedLatch.await(getTestTimeout(),
MILLISECONDS));
- discoveryRingMessageWorkerQueue(grid(0)).block();
+ blockingDiscovery(grid(0)).block();
IgniteInternalFuture<Object> finalizeFut =
GridTestUtils.runAsync(() -> finalizeClusterVersion(1, TEST_DEFAULT_VER));
@@ -880,7 +884,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
waitForBlockedDiscoveryMessages(grid(0), 1,
TcpDiscoveryNodeAddedMessage.class);
- discoveryRingMessageWorkerQueue(grid(0)).unblock();
+ blockingDiscovery(grid(0)).unblock();
finalizeFut.get(getTestTimeout(), MILLISECONDS);
@@ -894,13 +898,11 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
/** */
@Test
public void testConcurrentFinalizationAndNodeJoin() throws Exception {
- wrapRingMessageWorkerQueue(startGrid(0, TEST_DEFAULT_VER));
- startGrid(1, TEST_DEFAULT_VER);
- startClientGrid(2, TEST_DEFAULT_VER);
+ startCluster();
ru(1).enableVersionUpgrade();
- discoveryRingMessageWorkerQueue(grid(0)).block();
+ blockingDiscovery(grid(0)).block();
IgniteInternalFuture<Object> finalizeFut = GridTestUtils.runAsync(()
-> ru(1).finalizeClusterVersion());
@@ -910,7 +912,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
waitForBlockedDiscoveryMessages(grid(0), 1,
TcpDiscoveryJoinRequestMessage.class);
- discoveryRingMessageWorkerQueue(grid(0)).unblock();
+ blockingDiscovery(grid(0)).unblock();
finalizeFut.get(getTestTimeout(), MILLISECONDS);
@@ -935,7 +937,7 @@ public class CoreVersionRollingUpgradeTest extends
AbstractRollingUpgradeTest {
private boolean queuedDiscoveryMessageCountMatches(IgniteEx ignite, int
expCnt, Class<?> msgCls) {
int cnt = 0;
- for (TcpDiscoveryAbstractMessage msg :
discoveryRingMessageWorkerQueue(ignite)) {
+ for (TcpDiscoveryAbstractMessage msg :
blockingDiscovery(ignite).messageQueue()) {
Class<?> queuedMsgCls = (msg instanceof
TcpDiscoveryCustomEventMessage customMsg)
? customMsg.message().getClass()
: msg.getClass();
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/security/NodeSecurityContextPropagationTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/security/NodeSecurityContextPropagationTest.java
index 6e33bed6660..c0a4a503c71 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/security/NodeSecurityContextPropagationTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/security/NodeSecurityContextPropagationTest.java
@@ -20,14 +20,10 @@ package org.apache.ignite.internal.processors.security;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
-import java.util.Iterator;
import java.util.Objects;
import java.util.UUID;
-import java.util.concurrent.BlockingDeque;
import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
import java.util.function.Predicate;
-import org.apache.commons.lang3.reflect.FieldUtils;
import org.apache.ignite.configuration.DataRegionConfiguration;
import org.apache.ignite.configuration.DataStorageConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
@@ -37,22 +33,20 @@ import
org.apache.ignite.internal.events.DiscoveryCustomEvent;
import
org.apache.ignite.internal.managers.discovery.SecurityAwareCustomMessageWrapper;
import org.apache.ignite.internal.util.typedef.internal.U;
import org.apache.ignite.spi.MessagesPluginProvider;
-import org.apache.ignite.spi.discovery.DiscoverySpi;
import org.apache.ignite.spi.discovery.DiscoverySpiCustomMessage;
-import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.TestBlockingTcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryCustomEventMessage;
import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeAddedMessage;
import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryNodeLeftMessage;
import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
import org.junit.Test;
import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
import static
org.apache.ignite.internal.IgniteNodeAttributes.ATTR_IGNITE_INSTANCE_NAME;
import static
org.apache.ignite.internal.events.DiscoveryCustomEvent.EVT_DISCOVERY_CUSTOM_EVT;
+import static
org.apache.ignite.spi.discovery.tcp.TestBlockingTcpDiscoverySpi.blockingDiscovery;
import static org.apache.ignite.testframework.GridTestUtils.runAsync;
import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
@@ -81,9 +75,8 @@ public class NodeSecurityContextPropagationTest extends
GridCommonAbstractTest {
.setMaxSize(100L * 1024 * 1024)))
.setAuthenticationEnabled(true);
- ((TcpDiscoverySpi)cfg.getDiscoverySpi())
- .setIpFinder(new TcpDiscoveryVmIpFinder()
- .setAddresses(Collections.singleton("127.0.0.1:47500")));
+ cfg.setDiscoverySpi(new TestBlockingTcpDiscoverySpi(new
TcpDiscoveryVmIpFinder()
+ .setAddresses(Collections.singleton("127.0.0.1:47500"))));
cfg.setPluginProviders(new
MessagesPluginProvider(TestDiscoveryMessage.class,
TestDiscoveryAcknowledgeMessage.class));
@@ -104,15 +97,6 @@ public class NodeSecurityContextPropagationTest extends
GridCommonAbstractTest {
cleanPersistenceDir();
}
- /** {@inheritDoc} */
- @Override protected IgniteEx startGrid(int idx) throws Exception {
- IgniteEx ignite = super.startGrid(idx);
-
- wrapRingMessageWorkerQueue(ignite);
-
- return ignite;
- }
-
/** */
@Test
public void testProcessCustomDiscoveryMessageFromLeftNode() throws
Exception {
@@ -133,12 +117,7 @@ public class NodeSecurityContextPropagationTest extends
GridCommonAbstractTest {
EVT_NODE_LEFT
);
- discoveryRingMessageWorkerQueue(srv).block();
-
- long pollingTimeout = U.field(discoveryRingMessageWorker(srv),
"pollingTimeout");
-
- // We need to wait for any active BlockingDeque#poll operation to
complete.
- U.sleep(5 * pollingTimeout);
+ blockingDiscovery(srv).block();
cli.context().discovery().sendCustomEvent(new TestDiscoveryMessage());
@@ -158,7 +137,7 @@ public class NodeSecurityContextPropagationTest extends
GridCommonAbstractTest {
waitForCondition(() -> anyReceivedMessageMatch(srv, msg ->
isDiscoveryNodeAddedMessage(msg, 3)), getTestTimeout());
- discoveryRingMessageWorkerQueue(srv).unblock();
+ blockingDiscovery(srv).unblock();
waitForCondition(
() -> grid(0).cluster().nodes().size() == 4
@@ -173,14 +152,9 @@ public class NodeSecurityContextPropagationTest extends
GridCommonAbstractTest {
((TcpDiscoveryNodeAddedMessage)msg).node().attribute(ATTR_IGNITE_INSTANCE_NAME));
}
- /** */
- public static BlockingDequeWrapper<TcpDiscoveryAbstractMessage>
discoveryRingMessageWorkerQueue(IgniteEx ignite) {
- return U.field(discoveryRingMessageWorker(ignite), "queue");
- }
-
/** */
private boolean anyReceivedMessageMatch(IgniteEx ignite, Predicate<Object>
predicate) {
- for (TcpDiscoveryAbstractMessage msg :
discoveryRingMessageWorkerQueue(ignite)) {
+ for (TcpDiscoveryAbstractMessage msg :
blockingDiscovery(ignite).messageQueue()) {
Object unwrappedMsg = msg;
if (msg instanceof TcpDiscoveryCustomEventMessage) {
@@ -197,312 +171,4 @@ public class NodeSecurityContextPropagationTest extends
GridCommonAbstractTest {
return false;
}
-
- /** */
- public static void wrapRingMessageWorkerQueue(IgniteEx ignite) throws
Exception {
- Object discoMsgWorker = discoveryRingMessageWorker(ignite);
-
- BlockingDeque<TcpDiscoveryAbstractMessage> queue =
U.field(discoMsgWorker, "queue");
-
- BlockingDequeWrapper<TcpDiscoveryAbstractMessage> wrapper = new
BlockingDequeWrapper<>(queue);
-
- FieldUtils.writeField(discoMsgWorker, "queue", wrapper, true);
- }
-
- /** */
- public static Object discoveryRingMessageWorker(IgniteEx ignite) {
- DiscoverySpi[] discoverySpis = U.field(ignite.context().discovery(),
"spis");
-
- Object impl = U.field(discoverySpis[0], "impl");
-
- return U.field(impl, "msgWorker");
- }
-
- /** */
- public static class BlockingDequeWrapper<T> implements BlockingDeque<T> {
- /** */
- private volatile boolean isBlocked;
-
- /** */
- private final BlockingDeque<T> delegate;
-
- /** */
- private Predicate<T> msgInterceptor;
-
- /** */
- public BlockingDequeWrapper(BlockingDeque<T> delegate) {
- this.delegate = delegate;
- }
-
- /** */
- public void block() {
- isBlocked = true;
- }
-
- /** */
- public void unblock() {
- isBlocked = false;
- }
-
- /** */
- public void startMessageIntercepting(Predicate<T> msgInterceptor) {
- this.msgInterceptor = msgInterceptor;
- }
-
- /** {@inheritDoc} */
- @Override public T poll(long timeout, TimeUnit unit) throws
InterruptedException {
- return isBlocked ? null : delegate.poll(timeout, unit);
- }
-
- /** {@inheritDoc} */
- @Override public int remainingCapacity() {
- return delegate.remainingCapacity();
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public T element() {
- return delegate.element();
- }
-
- /** {@inheritDoc} */
- @Override public T peek() {
- return delegate.peek();
- }
-
- /** {@inheritDoc} */
- @Override public boolean remove(Object o) {
- return delegate.remove(o);
- }
-
- /** {@inheritDoc} */
- @Override public boolean containsAll(@NotNull Collection<?> c) {
- return delegate.containsAll(c);
- }
-
- /** {@inheritDoc} */
- @Override public boolean addAll(@NotNull Collection<? extends T> c) {
- return delegate.addAll(c);
- }
-
- /** {@inheritDoc} */
- @Override public boolean removeAll(@NotNull Collection<?> c) {
- return delegate.removeAll(c);
- }
-
- /** {@inheritDoc} */
- @Override public boolean retainAll(@NotNull Collection<?> c) {
- return delegate.retainAll(c);
- }
-
- /** {@inheritDoc} */
- @Override public void clear() {
- delegate.clear();
- }
-
- /** {@inheritDoc} */
- @Override public void addFirst(T t) {
- if (msgInterceptor != null && !msgInterceptor.test(t))
- return;
-
- delegate.addFirst(t);
- }
-
- /** {@inheritDoc} */
- @Override public void addLast(@NotNull T t) {
- delegate.addLast(t);
- }
-
- /** {@inheritDoc} */
- @Override public boolean offerFirst(@NotNull T t) {
- return delegate.offerFirst(t);
- }
-
- /** {@inheritDoc} */
- @Override public boolean offerLast(@NotNull T t) {
- return delegate.offerLast(t);
- }
-
- /** {@inheritDoc} */
- @Override public T removeFirst() {
- return delegate.removeFirst();
- }
-
- /** {@inheritDoc} */
- @Override public T removeLast() {
- return delegate.removeLast();
- }
-
- /** {@inheritDoc} */
- @Override public T pollFirst() {
- return delegate.pollFirst();
- }
-
- /** {@inheritDoc} */
- @Override public T pollLast() {
- return delegate.pollLast();
- }
-
- /** {@inheritDoc} */
- @Override public T getFirst() {
- return delegate.getFirst();
- }
-
- /** {@inheritDoc} */
- @Override public T getLast() {
- return delegate.getLast();
- }
-
- /** {@inheritDoc} */
- @Override public T peekFirst() {
- return delegate.peekFirst();
- }
-
- /** {@inheritDoc} */
- @Override public T peekLast() {
- return delegate.peekLast();
- }
-
- /** {@inheritDoc} */
- @Override public void putFirst(@NotNull T t) throws
InterruptedException {
- delegate.putFirst(t);
- }
-
- /** {@inheritDoc} */
- @Override public void putLast(@NotNull T t) throws
InterruptedException {
- delegate.putLast(t);
- }
-
- /** {@inheritDoc} */
- @Override public boolean offerFirst(@NotNull T t, long timeout,
TimeUnit unit) throws InterruptedException {
- return delegate.offerFirst(t, timeout, unit);
- }
-
- /** {@inheritDoc} */
- @Override public boolean offerLast(@NotNull T t, long timeout,
TimeUnit unit) throws InterruptedException {
- return delegate.offerLast(t, timeout, unit);
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public T takeFirst() throws InterruptedException {
- return delegate.takeFirst();
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public T takeLast() throws InterruptedException {
- return delegate.takeLast();
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public T pollFirst(long timeout, TimeUnit unit)
throws InterruptedException {
- return delegate.pollFirst(timeout, unit);
- }
-
- /** {@inheritDoc} */
- @Nullable @Override public T pollLast(long timeout, TimeUnit unit)
throws InterruptedException {
- return delegate.pollLast(timeout, unit);
- }
-
- /** {@inheritDoc} */
- @Override public boolean removeFirstOccurrence(Object o) {
- return delegate.removeFirstOccurrence(o);
- }
-
- /** {@inheritDoc} */
- @Override public boolean removeLastOccurrence(Object o) {
- return delegate.removeLastOccurrence(o);
- }
-
- /** {@inheritDoc} */
- @Override public boolean add(T t) {
- if (msgInterceptor != null && !msgInterceptor.test(t))
- return true;
-
- return delegate.add(t);
- }
-
- /** {@inheritDoc} */
- @Override public boolean offer(@NotNull T t) {
- return delegate.offer(t);
- }
-
- /** {@inheritDoc} */
- @Override public void put(@NotNull T t) throws InterruptedException {
- delegate.put(t);
- }
-
- /** {@inheritDoc} */
- @Override public boolean offer(@NotNull T t, long timeout, TimeUnit
unit) throws InterruptedException {
- return delegate.offer(t, timeout, unit);
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public T remove() {
- return delegate.remove();
- }
-
- /** {@inheritDoc} */
- @Override public T poll() {
- return delegate.poll();
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public T take() throws InterruptedException {
- return delegate.take();
- }
-
- /** {@inheritDoc} */
- @Override public boolean contains(Object o) {
- return delegate.contains(o);
- }
-
- /** {@inheritDoc} */
- @Override public int drainTo(@NotNull Collection<? super T> c) {
- return delegate.drainTo(c);
- }
-
- /** {@inheritDoc} */
- @Override public int drainTo(@NotNull Collection<? super T> c, int
maxElements) {
- return delegate.drainTo(c, maxElements);
- }
-
- /** {@inheritDoc} */
- @Override public int size() {
- return delegate.size();
- }
-
- /** {@inheritDoc} */
- @Override public boolean isEmpty() {
- return delegate.isEmpty();
- }
-
- /** {@inheritDoc} */
- @Override public Iterator<T> iterator() {
- return delegate.iterator();
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public Object[] toArray() {
- return delegate.toArray();
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public <T1> T1[] toArray(@NotNull T1[] a) {
- return delegate.toArray(a);
- }
-
- /** {@inheritDoc} */
- @NotNull @Override public Iterator<T> descendingIterator() {
- return delegate.descendingIterator();
- }
-
- /** {@inheritDoc} */
- @Override public void push(@NotNull T t) {
- delegate.push(t);
- }
-
- /** {@inheritDoc} */
- @Override public T pop() {
- return delegate.pop();
- }
- }
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/MultiDataCenterSplitTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/MultiDataCenterSplitTest.java
index b2d45216d38..73acc8b9a37 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/MultiDataCenterSplitTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/MultiDataCenterSplitTest.java
@@ -338,15 +338,8 @@ public class MultiDataCenterSplitTest extends
GridCommonAbstractTest {
}
/** {@inheritDoc} */
- @Override protected void initializeImpl() {
- if (impl != null)
- return;
-
- super.initializeImpl();
-
- // In theory, might be a ClientImpl.
- if (impl instanceof ServerImpl)
- impl = new ServerImpl(this, DFLT_UTLITY_POOL_SIZE,
pingPoolSize);
+ @Override TcpDiscoveryImpl createServerTcpDiscoveryImplementation() {
+ return new ServerImpl(this, DFLT_UTLITY_POOL_SIZE, pingPoolSize);
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryNetworkIssuesTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryNetworkIssuesTest.java
index f87a0753891..f416144d49b 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryNetworkIssuesTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryNetworkIssuesTest.java
@@ -593,16 +593,10 @@ public class TcpDiscoveryNetworkIssuesTest extends
GridCommonAbstractTest {
private final AtomicReference<Collection<InetSocketAddress>>
simulatedPrevNodeAddr = new AtomicReference<>();
/** {@inheritDoc} */
- @Override protected void initializeImpl() {
- if (impl != null)
- return;
-
- super.initializeImpl();
-
- // To make the test stable, we want a loopback paddress of the
previous node responds first.
+ @Override TcpDiscoveryImpl createServerTcpDiscoveryImplementation() {
+ // To make the test stable, we want a loopback address of the
previous node responds first.
// We don't need a concurrent ping execution.
- if (impl instanceof ServerImpl)
- impl = new ServerImpl(this, 1, DFLT_RMT_DC_PING_POOL_SIZE);
+ return new ServerImpl(this, 1, DFLT_RMT_DC_PING_POOL_SIZE);
}
/** */
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryPendingMessageDeliveryTest.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryPendingMessageDeliveryTest.java
index bdd67ce404f..503d1198e5f 100644
---
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryPendingMessageDeliveryTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoveryPendingMessageDeliveryTest.java
@@ -238,26 +238,17 @@ public class TcpDiscoveryPendingMessageDeliveryTest
extends GridCommonAbstractTe
/** {@link TcpDiscoverySpi} which doesn't itsring to current DC. */
private class NoRingClosingTcpDiscoverySpi extends TestTcpDiscoverySpi {
/** {@inheritDoc} */
- @Override protected void initializeImpl() {
- if (impl != null)
- return;
-
- super.initializeImpl();
-
- // In theory, might be a ClientImpl.
- if (impl instanceof ServerImpl) {
- impl = new ServerImpl(this, DFLT_UTLITY_POOL_SIZE, 0) {
- @Override protected ServerImpl.RingMessageWorker
createMessageWorker() {
- return new ServerImpl.RingMessageWorker(impl.log) {
- @Override protected
ServerImpl.CrossRingMessageSendState createConnectionRecoveryState(
- TcpDiscoveryNode n) {
- // Do not start remote DC ping.
- return new
ServerImpl.CrossRingMessageSendState();
- }
- };
- }
- };
- }
+ @Override TcpDiscoveryImpl createServerTcpDiscoveryImplementation() {
+ return new ServerImpl(this, DFLT_UTLITY_POOL_SIZE, 0) {
+ @Override protected ServerImpl.RingMessageWorker
createMessageWorker() {
+ return new ServerImpl.RingMessageWorker(log) {
+ @Override protected
ServerImpl.CrossRingMessageSendState
createConnectionRecoveryState(TcpDiscoveryNode n) {
+ // Do not start remote DC ping.
+ return new ServerImpl.CrossRingMessageSendState();
+ }
+ };
+ }
+ };
}
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestBlockingTcpDiscoverySpi.java
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestBlockingTcpDiscoverySpi.java
new file mode 100644
index 00000000000..834191722f6
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/spi/discovery/tcp/TestBlockingTcpDiscoverySpi.java
@@ -0,0 +1,122 @@
+/*
+ * 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.ignite.spi.discovery.tcp;
+
+import java.util.concurrent.BlockingDeque;
+import java.util.concurrent.LinkedBlockingDeque;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Predicate;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import
org.apache.ignite.spi.discovery.tcp.messages.TcpDiscoveryAbstractMessage;
+import org.jspecify.annotations.NonNull;
+
+/** */
+public class TestBlockingTcpDiscoverySpi extends TestTcpDiscoverySpi {
+ /** */
+ private final BlockingDeque<TcpDiscoveryAbstractMessage> msgQueue = new
TestBlockingLinkedDeque();
+
+ /** */
+ private volatile boolean isBlocked;
+
+ /** */
+ private volatile Predicate<TcpDiscoveryAbstractMessage> msgFilter;
+
+ /** */
+ public TestBlockingTcpDiscoverySpi(TcpDiscoveryIpFinder ipFinder) {
+ setIpFinder(ipFinder);
+ }
+
+ /** */
+ public TestBlockingTcpDiscoverySpi(IgniteConfiguration cfg) {
+ this(((TcpDiscoverySpi)cfg.getDiscoverySpi()).getIpFinder());
+ }
+
+ /** {@inheritDoc} */
+ @Override TcpDiscoveryImpl createServerTcpDiscoveryImplementation() {
+ return new ServerImpl(this, DFLT_UTLITY_POOL_SIZE,
DFLT_RMT_DC_PING_POOL_SIZE) {
+ @Override protected ServerImpl.RingMessageWorker
createMessageWorker() {
+ return new RingMessageWorker(log, msgQueue);
+ }
+ };
+ }
+
+ /** */
+ public void block() {
+ isBlocked = true;
+ }
+
+ /** */
+ public void unblock() {
+ isBlocked = false;
+ }
+
+ /** */
+ public void messageFilter(Predicate<TcpDiscoveryAbstractMessage>
msgFilter) {
+ this.msgFilter = msgFilter;
+ }
+
+ /** */
+ public BlockingDeque<TcpDiscoveryAbstractMessage> messageQueue() {
+ return msgQueue;
+ }
+
+ /** */
+ public static TestBlockingTcpDiscoverySpi blockingDiscovery(Ignite node) {
+ assert !node.configuration().isClientMode();
+
+ return
(TestBlockingTcpDiscoverySpi)node.configuration().getDiscoverySpi();
+ }
+
+ /** */
+ private class TestBlockingLinkedDeque extends
LinkedBlockingDeque<TcpDiscoveryAbstractMessage> {
+ /** {@inheritDoc} */
+ @Override public TcpDiscoveryAbstractMessage poll(long timeout,
TimeUnit unit) throws InterruptedException {
+ if (isBlocked)
+ return null;
+
+ TcpDiscoveryAbstractMessage msg = super.poll(timeout, unit);
+
+ if (isBlocked && msg != null) {
+ // Since this queue is processed by a single thread, it is
safe to simply put the element back.
+ super.addFirst(msg);
+
+ return null;
+ }
+
+ return msg;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void addFirst(@NonNull TcpDiscoveryAbstractMessage t)
{
+ if (msgFilter != null && msgFilter.test(t))
+ return;
+
+ super.addFirst(t);
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean add(@NonNull TcpDiscoveryAbstractMessage t) {
+ if (msgFilter != null && msgFilter.test(t))
+ return true;
+
+ return super.add(t);
+ }
+ }
+}
