This is an automated email from the ASF dual-hosted git repository.
nizhikov 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 0a612438b3d IGNITE-24098 Remove PROTO COL_V2_VERSION_SINCE (#13394)
0a612438b3d is described below
commit 0a612438b3dc603ed69d4717ef0536adb90eb7cb
Author: Nikolay <[email protected]>
AuthorDate: Wed Jul 22 21:06:36 2026 +0300
IGNITE-24098 Remove PROTO COL_V2_VERSION_SINCE (#13394)
---
.../preloader/GridDhtPartitionsExchangeFuture.java | 7 +-
.../dht/preloader/latch/ExchangeLatchManager.java | 30 +--
...IgniteExchangeLatchManagerDiscoHistoryTest.java | 296 ---------------------
...IgniteExchangeLatchManagerDiscoHistoryTest.java | 71 +++++
.../ignite/testsuites/IgniteCacheTestSuite16.java | 2 +-
5 files changed, 76 insertions(+), 330 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
index e3c7b42dc7f..473508fba29 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java
@@ -1617,8 +1617,7 @@ public class GridDhtPartitionsExchangeFuture extends
GridDhtTopologyFutureAdapte
timeBag.finishGlobalStage("Preloading notification");
// Skipping wait on local join is available when all cluster nodes
have the same protocol.
- boolean skipWaitOnLocJoin = localJoinExchange()
- && cctx.exchange().latch().canSkipJoiningNodes(initialVersion());
+ boolean skipWaitOnLocJoin = localJoinExchange();
if (context().exchangeFreeSwitch() && isBaselineNodeFailed())
waitPartitionRelease(null, false, false);
@@ -1970,10 +1969,6 @@ public class GridDhtPartitionsExchangeFuture extends
GridDhtTopologyFutureAdapte
releaseLatch.countDown();
- // For compatibility with old version where joining nodes are not
waiting for latch.
- if (localJoinExchange() &&
!cctx.exchange().latch().canSkipJoiningNodes(initialVersion()))
- return;
-
try {
String troubleshootingHint;
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/latch/ExchangeLatchManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/latch/ExchangeLatchManager.java
index 4d03edce1fc..ef9a63ec129 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/latch/ExchangeLatchManager.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/latch/ExchangeLatchManager.java
@@ -48,7 +48,6 @@ import org.apache.ignite.internal.util.lang.GridPlainRunnable;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.internal.S;
-import org.apache.ignite.lang.IgniteProductVersion;
import org.jetbrains.annotations.Nullable;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
@@ -59,12 +58,6 @@ import static
org.apache.ignite.events.EventType.EVT_NODE_LEFT;
* Class is responsible to create and manage instances of distributed latches
{@link Latch}.
*/
public class ExchangeLatchManager {
- /**
- * Exchange latch V2 protocol introduces following optimization: Joining
nodes are explicitly excluded from possible
- * latch participants.
- */
- public static final IgniteProductVersion PROTOCOL_V2_VERSION_SINCE =
IgniteProductVersion.fromString("2.5.3");
-
/** Logger. */
private final IgniteLogger log;
@@ -275,7 +268,7 @@ public class ExchangeLatchManager {
* @return Collection of nodes with at least one cache configured.
* @throws IgniteException If nodes for the given {@code topVer} cannot be
found in the discovery history.
*/
- private Collection<ClusterNode>
aliveNodesForTopologyVer(AffinityTopologyVersion topVer) {
+ Collection<ClusterNode> aliveNodesForTopologyVer(AffinityTopologyVersion
topVer) {
if (topVer == AffinityTopologyVersion.NONE)
return discovery.aliveServerNodes();
else {
@@ -298,10 +291,7 @@ public class ExchangeLatchManager {
private Collection<ClusterNode>
getLatchParticipants(AffinityTopologyVersion topVer) {
List<ClusterNode> participantNodes = new
ArrayList<>(aliveNodesForTopologyVer(topVer));
- if (canSkipJoiningNodes(topVer))
- return excludeJoinedNodes(participantNodes, topVer);
-
- return participantNodes;
+ return excludeJoinedNodes(participantNodes, topVer);
}
/**
@@ -334,25 +324,11 @@ public class ExchangeLatchManager {
if (applicableNodes.isEmpty())
return null;
- if (canSkipJoiningNodes(topVer))
- applicableNodes = excludeJoinedNodes(applicableNodes, topVer);
+ applicableNodes = excludeJoinedNodes(applicableNodes, topVer);
return applicableNodes.get(0);
}
- /**
- * Checks that latch manager can use V2 protocol and skip joining nodes
from latch participants.
- *
- * @param topVer Topology version.
- * @throws IgniteException If nodes for the given {@code topVer} cannot be
found in the discovery history.
- */
- public boolean canSkipJoiningNodes(AffinityTopologyVersion topVer) {
- Collection<ClusterNode> applicableNodes =
aliveNodesForTopologyVer(topVer);
-
- return applicableNodes.stream()
- .allMatch(node ->
node.version().compareTo(PROTOCOL_V2_VERSION_SINCE) >= 0);
- }
-
/**
* Processes ack message from given {@code from} node.
*
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteExchangeLatchManagerDiscoHistoryTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteExchangeLatchManagerDiscoHistoryTest.java
deleted file mode 100644
index d7bde459fa5..00000000000
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/datastructures/IgniteExchangeLatchManagerDiscoHistoryTest.java
+++ /dev/null
@@ -1,296 +0,0 @@
-/*
- * 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.internal.processors.cache.datastructures;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicReference;
-import org.apache.ignite.Ignite;
-import org.apache.ignite.IgniteCheckedException;
-import org.apache.ignite.IgniteException;
-import org.apache.ignite.ShutdownPolicy;
-import org.apache.ignite.configuration.IgniteConfiguration;
-import org.apache.ignite.failure.AbstractFailureHandler;
-import org.apache.ignite.failure.FailureContext;
-import org.apache.ignite.internal.IgniteEx;
-import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.IgnitionEx;
-import
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
-import
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.PartitionsExchangeAware;
-import
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.latch.ExchangeLatchManager;
-import
org.apache.ignite.internal.processors.metastorage.DistributedMetastorageLifecycleListener;
-import
org.apache.ignite.internal.processors.metastorage.ReadableDistributedMetaStorage;
-import org.apache.ignite.internal.util.typedef.X;
-import org.apache.ignite.lifecycle.LifecycleBean;
-import org.apache.ignite.lifecycle.LifecycleEventType;
-import org.apache.ignite.resources.IgniteInstanceResource;
-import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
-import org.apache.ignite.spi.discovery.tcp.TestTcpDiscoverySpi;
-import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
-import org.apache.ignite.testframework.GridTestUtils;
-import org.apache.ignite.testframework.junits.WithSystemProperty;
-import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
-import org.junit.After;
-import org.junit.Before;
-import org.junit.Test;
-
-import static
org.apache.ignite.IgniteSystemProperties.IGNITE_DISCOVERY_HISTORY_SIZE;
-
-/**
- * Test {@link ExchangeLatchManager} throws {@link IgniteException} with
appropriate message when topology history
- * cannot be obtained.
- */
-public class IgniteExchangeLatchManagerDiscoHistoryTest extends
GridCommonAbstractTest {
- /** Topology history size. */
- private static final int TOPOLOGY_HISTORY_SIZE = 2;
-
- /** Disco cache size. */
- private static final String DISCO_HISTORY_SIZE = "2";
-
- /** Timeout in millis. */
- private static final long DEFAULT_TIMEOUT = 30_000;
-
- /** Lifecycle bean that is used to register required listeners. */
- private LifecycleBean lifecycleBean;
-
- /** Flag indicates that a node, that is starting, should have short
topology history. */
- private boolean victim;
-
- /** Discovery SPI that is used by the node with short topology history. */
- private CustomTcpDiscoverySpi disco;
-
- /** Failure context. */
- private final AtomicReference<FailureContext> cpFailureCtx = new
AtomicReference<>();
-
- /** {@inheritDoc} */
- @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
- IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
-
- TcpDiscoveryIpFinder ipFinder =
((TcpDiscoverySpi)cfg.getDiscoverySpi()).getIpFinder();
-
- int topHistSize = victim ? TOPOLOGY_HISTORY_SIZE :
TcpDiscoverySpi.DFLT_TOP_HISTORY_SIZE;
-
- CustomTcpDiscoverySpi discoSpi = new
CustomTcpDiscoverySpi(topHistSize, ipFinder);
-
- cfg.setDiscoverySpi(discoSpi);
-
- if (victim) {
- cfg.setFailureHandler(new AbstractFailureHandler() {
- /** {@inheritDoc} */
- @Override protected boolean handle(Ignite ignite,
FailureContext failureCtx) {
- cpFailureCtx.compareAndSet(null, failureCtx);
-
- // Invalidate kernel context.
- return true;
- }
- });
-
- cfg.setLifecycleBeans(lifecycleBean);
-
- disco = discoSpi;
- }
-
- return cfg;
- }
-
- /**
- * Prepares test for execution.
- */
- @Before
- public void startup() {
- shutdown();
- }
-
- /**
- * Cleans after the test.
- */
- @After
- public void shutdown() {
- lifecycleBean = null;
-
- victim = false;
-
- cpFailureCtx.set(null);
-
- disco = null;
-
- stopAllGrids();
- }
-
- /**
- * @throws Exception If failed.
- */
- @Test
- @WithSystemProperty(key = IGNITE_DISCOVERY_HISTORY_SIZE, value =
DISCO_HISTORY_SIZE)
- public void testProperException() throws Exception {
- final IgniteEx crd = startGrid(0);
-
- final CountDownLatch exchangeLatch = new CountDownLatch(1);
-
- final CountDownLatch startSrvsLatch = new CountDownLatch(1);
-
- final AtomicReference<Exception> err = new AtomicReference<>();
-
- // Lifecycle bean that is used to register PartitionsExchangeAware
listener.
- lifecycleBean = new LifecycleBean() {
- /** Ignite instance. */
- @IgniteInstanceResource
- IgniteEx ignite;
-
- /** {@inheritDoc} */
- @Override public void onLifecycleEvent(LifecycleEventType evt)
throws IgniteException {
- if (evt == LifecycleEventType.BEFORE_NODE_START) {
- // The goal is registering PartitionsExchangeAware
listener before the discovery manager is started.
- DistributedMetastorageLifecycleListener lsnr = new
DistributedMetastorageLifecycleListener() {
- @Override public void
onReadyForRead(ReadableDistributedMetaStorage metastorage) {
- ignite.context().cache().context().exchange()
- .registerExchangeAwareComponent(new
PartitionsExchangeAware() {
- /** {@inheritDoc} */
- @Override public void
onInitBeforeTopologyLock(
- GridDhtPartitionsExchangeFuture fut) {
- try {
- // Let's start nodes.
- startSrvsLatch.countDown();
-
- // Blocks the initial exchange and
waits for other nodes.
-
exchangeLatch.await(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
- }
- catch (Exception e) {
- err.compareAndSet(null, e);
- }
- }
- });
- }
- };
-
-
ignite.context().internalSubscriptionProcessor().registerDistributedMetastorageListener(lsnr);
- }
- }
- };
-
- // Start server node with short topology history.
- victim = true;
-
- GridTestUtils.runAsync(() -> startGrid(1));
-
- // Waits for the initial exchange.
- startSrvsLatch.await(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
-
- victim = false;
-
- lifecycleBean = null;
-
- List<IgniteInternalFuture> srvFuts = new
ArrayList<>(TOPOLOGY_HISTORY_SIZE);
-
- try {
- // Major topology version that is corresponding to the start of
the node with short topology history.
- final long topVer = 2;
-
- // Starting server nodes to exhaust the topology history.
- for (int i = 2; i < 3 * TOPOLOGY_HISTORY_SIZE &&
!disco.isEmptyTopologyHistory(topVer); ++i) {
- final int currNodeIdx = i;
-
- final int joinedNodesCnt = disco.totalJoinedNodes();
-
- srvFuts.add(GridTestUtils.runAsync(() ->
startGrid(currNodeIdx)));
-
- assertTrue("Failed to wait for a new server node
[joinedNodesCnt=" + joinedNodesCnt + "]",
- GridTestUtils.waitForCondition(
- () -> disco.totalJoinedNodes() >= (joinedNodesCnt +
1), DEFAULT_TIMEOUT));
- }
-
- assertTrue(
- "Disco cache history is not empty for the topology
[majorTopVer=" + topVer + ']',
- disco.isEmptyTopologyHistory(topVer));
-
- // Let's continue the ongoing exchange.
- exchangeLatch.countDown();
-
- boolean failureHnd = GridTestUtils.waitForCondition(() ->
cpFailureCtx.get() != null, DEFAULT_TIMEOUT);
-
- assertNull(
- "Unexpected exception (probably, the topology history still
exists [err=" + err + ']',
- err.get());
-
- assertTrue("Failure handler was not triggered.", failureHnd);
-
- // Check that IgniteException was thrown instead of
NullPointerException.
- assertTrue(
- "IgniteException must be thrown.",
- X.hasCause(cpFailureCtx.get().error(), IgniteException.class));
-
- // Check that message contains a hint to fix the issue.
- GridTestUtils.assertContains(
- log,
- cpFailureCtx.get().error().getMessage(),
- "Consider increasing IGNITE_DISCOVERY_HISTORY_SIZE property.
Current value is " + DISCO_HISTORY_SIZE);
- }
- finally {
- IgnitionEx.stop(getTestIgniteInstanceName(1), true,
ShutdownPolicy.IMMEDIATE, true);
-
- srvFuts.forEach(f -> {
- try {
- f.get(DEFAULT_TIMEOUT);
- }
- catch (IgniteCheckedException e) {
- err.compareAndSet(null, e);
- }
- });
- }
-
- assertNull("Unexpected exception [err=" + err.get() + ']', err.get());
- }
-
- /**
- * Custom discovery SPI that allows specifying a short topology history
for testing purposes.
- */
- private static class CustomTcpDiscoverySpi extends TestTcpDiscoverySpi {
- /**
- * Creates a new instance of discovery spi with the given size of
topology snapshots history.
- *
- * @param topHistSize Size of topology snapshots history.
- * @param ipFinder IP finder
- */
- CustomTcpDiscoverySpi(int topHistSize, TcpDiscoveryIpFinder ipFinder) {
- this.topHistSize = topHistSize;
-
- setIpFinder(ipFinder);
- }
-
- /**
- * Returns the number of joined nodes.
- *
- * @return Number of joined nodes.
- */
- int totalJoinedNodes() {
- return stats.joinedNodesCount();
- }
-
- /**
- * Returns {@code true} if the given topology version is already
removed from the history.
- *
- * @param topVer Major topology version.
- * @return {@code true} if the given topology version is already
removed from the history.
- */
- boolean isEmptyTopologyHistory(long topVer) {
- return ((IgniteEx)ignite).context().discovery().topology(topVer)
== null;
- }
- }
-}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/latch/IgniteExchangeLatchManagerDiscoHistoryTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/latch/IgniteExchangeLatchManagerDiscoHistoryTest.java
new file mode 100644
index 00000000000..ef48cb99af1
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/latch/IgniteExchangeLatchManagerDiscoHistoryTest.java
@@ -0,0 +1,71 @@
+/*
+ * 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.internal.processors.cache.distributed.dht.preloader.latch;
+
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.TestTcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.testframework.junits.WithSystemProperty;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Test;
+
+import static
org.apache.ignite.IgniteSystemProperties.IGNITE_DISCOVERY_HISTORY_SIZE;
+import static org.apache.ignite.testframework.GridTestUtils.assertThrows;
+
+/** */
+public class IgniteExchangeLatchManagerDiscoHistoryTest extends
GridCommonAbstractTest {
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName);
+
+ TcpDiscoveryIpFinder ipFinder =
((TcpDiscoverySpi)cfg.getDiscoverySpi()).getIpFinder();
+
+ int topHistSize = 2;
+
+ return cfg.setDiscoverySpi(new
CustomTcpDiscoverySpi(topHistSize).setIpFinder(ipFinder));
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ @WithSystemProperty(key = IGNITE_DISCOVERY_HISTORY_SIZE, value = "2")
+ public void testProperException2() throws Exception {
+ IgniteEx srv = startGrids(3);
+
+ awaitPartitionMapExchange();
+
+ assertThrows(
+ null,
+ () ->
srv.context().cache().context().exchange().latch().aliveNodesForTopologyVer(new
AffinityTopologyVersion(1, 0)),
+ IgniteException.class,
+ "Consider increasing IGNITE_DISCOVERY_HISTORY_SIZE property.
Current value is " + 2
+ );
+ }
+
+ /** */
+ private static class CustomTcpDiscoverySpi extends TestTcpDiscoverySpi {
+ /** */
+ CustomTcpDiscoverySpi(int topHistSize) {
+ this.topHistSize = topHistSize;
+ }
+ }
+}
diff --git
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite16.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite16.java
index 2171ab5a1d1..ae9b6461040 100644
---
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite16.java
+++
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite16.java
@@ -24,13 +24,13 @@ import
org.apache.ignite.internal.processors.cache.PartitionedTransactionalOptim
import
org.apache.ignite.internal.processors.cache.ReplicatedAtomicCacheGetsDistributionTest;
import
org.apache.ignite.internal.processors.cache.ReplicatedTransactionalOptimisticCacheGetsDistributionTest;
import
org.apache.ignite.internal.processors.cache.ReplicatedTransactionalPessimisticCacheGetsDistributionTest;
-import
org.apache.ignite.internal.processors.cache.datastructures.IgniteExchangeLatchManagerDiscoHistoryTest;
import
org.apache.ignite.internal.processors.cache.distributed.CacheExchangeMergeMdcTest;
import
org.apache.ignite.internal.processors.cache.distributed.CacheParallelStartTest;
import
org.apache.ignite.internal.processors.cache.distributed.CacheTryLockMultithreadedTest;
import
org.apache.ignite.internal.processors.cache.distributed.IgniteCacheMultiClientsStartTest;
import
org.apache.ignite.internal.processors.cache.distributed.IgnitePessimisticTxSuspendResumeTest;
import
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.latch.ExchangeLatchManagerTest;
+import
org.apache.ignite.internal.processors.cache.distributed.dht.preloader.latch.IgniteExchangeLatchManagerDiscoHistoryTest;
import
org.apache.ignite.internal.processors.cache.transactions.TxMultiCacheAsyncOpsTest;
import
org.apache.ignite.internal.processors.cache.transactions.TxOnCachesStartTest;
import
org.apache.ignite.internal.processors.cache.transactions.TxOnCachesStopTest;
