You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by ir...@apache.org on 2020/02/07 15:50:59 UTC
[ignite] branch master updated: IGNITE-12636 Full rebalance instead
of a historical one - Fixes #7379.
This is an automated email from the ASF dual-hosted git repository.
irakov 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 67ac1d5 IGNITE-12636 Full rebalance instead of a historical one - Fixes #7379.
67ac1d5 is described below
commit 67ac1d5d68a2e2d13d5b3c361a8ab10b1d7873dc
Author: Mirza Aliev <al...@gmail.com>
AuthorDate: Fri Feb 7 18:48:14 2020 +0300
IGNITE-12636 Full rebalance instead of a historical one - Fixes #7379.
Signed-off-by: Ivan Rakov <ir...@apache.org>
---
.../cache/GridCachePartitionExchangeManager.java | 9 +-
.../dht/preloader/GridDhtPreloader.java | 2 +-
.../db/FullHistRebalanceOnClientStopTest.java | 257 +++++++++++++++++++++
.../ignite/testsuites/IgnitePdsTestSuite2.java | 3 +
4 files changed, 269 insertions(+), 2 deletions(-)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
index 308e6a4..e78ada3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCachePartitionExchangeManager.java
@@ -1117,6 +1117,13 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
}
/**
+ * @return {@code True} if pending future queue contains server exchange task.
+ */
+ public boolean hasPendingServerExchange() {
+ return exchWorker.hasPendingServerExchange();
+ }
+
+ /**
*
* @param topVer Topology version.
* @return Last topology version before the provided one when affinity was modified.
@@ -3437,7 +3444,7 @@ public class GridCachePartitionExchangeManager<K, V> extends GridCacheSharedMana
if (forcedRebFut != null)
forcedRebFut.markInitialized();
- if (assignsCancelled || hasPendingExchange()) {
+ if (assignsCancelled || hasPendingServerExchange()) {
U.log(log, "Skipping rebalancing (obsolete exchange ID) " +
"[top=" + resVer + ", evt=" + exchId.discoveryEventName() +
", node=" + exchId.nodeId() + ']');
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
index 2ab4d90..b8d3ff3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPreloader.java
@@ -202,7 +202,7 @@ public class GridDhtPreloader extends GridCachePreloaderAdapter {
CachePartitionFullCountersMap countersMap = grp.topology().fullUpdateCounters();
for (int p = 0; p < partitions; p++) {
- if (ctx.exchange().hasPendingExchange()) {
+ if (ctx.exchange().hasPendingServerExchange()) {
if (log.isDebugEnabled())
log.debug("Skipping assignments creation, exchange worker has pending assignments: " +
exchId);
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/FullHistRebalanceOnClientStopTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/FullHistRebalanceOnClientStopTest.java
new file mode 100644
index 0000000..9eaafd6
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/persistence/db/FullHistRebalanceOnClientStopTest.java
@@ -0,0 +1,257 @@
+/*
+ * 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.persistence.db;
+
+import java.util.concurrent.CountDownLatch;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction;
+import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.WALMode;
+import org.apache.ignite.internal.IgniteEx;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.TestRecordingCommunicationSpi;
+import org.apache.ignite.internal.managers.communication.GridIoMessage;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionDemandMessage;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.IgniteDhtDemandedPartitionsMap;
+import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.PartitionsExchangeAware;
+import org.apache.ignite.lang.IgniteInClosure;
+import org.apache.ignite.plugin.extensions.communication.Message;
+import org.apache.ignite.spi.IgniteSpiException;
+import org.apache.ignite.testframework.GridTestUtils;
+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_PDS_WAL_REBALANCE_THRESHOLD;
+
+/**
+ * Tests that full rebalance is not triggered instead of historical one when client node stops during PME
+ */
+public class FullHistRebalanceOnClientStopTest extends GridCommonAbstractTest {
+ /** Cache name. */
+ private static final String CACHE_NAME = "cache";
+
+ /** Partitions count. */
+ private static final int PARTS_CNT = 16;
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception {
+ IgniteConfiguration cfg = super.getConfiguration(gridName);
+
+ cfg.setConsistentId(gridName);
+
+ CacheConfiguration<Object, Object> ccfg1 = new CacheConfiguration<>(CACHE_NAME)
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ .setCacheMode(CacheMode.PARTITIONED)
+ .setBackups(1)
+ .setAffinity(new RendezvousAffinityFunction(false, PARTS_CNT));
+
+ cfg.setCacheConfiguration(ccfg1);
+
+ DataStorageConfiguration dbCfg = new DataStorageConfiguration()
+ .setWalMode(WALMode.LOG_ONLY)
+ .setDefaultDataRegionConfiguration(
+ new DataRegionConfiguration()
+ .setPersistenceEnabled(true)
+ .setMaxSize(100 * 1024 * 1024)
+ );
+
+ cfg.setDataStorageConfiguration(dbCfg);
+
+ cfg.setCommunicationSpi(new RebalanceCheckingCommunicationSpi());
+
+ return cfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ stopAllGrids();
+
+ cleanPersistenceDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ stopAllGrids();
+
+ cleanPersistenceDir();
+ }
+
+ /**
+ * @throws Exception if failed.
+ */
+ @Test
+ @WithSystemProperty(key = IGNITE_PDS_WAL_REBALANCE_THRESHOLD, value = "0")
+ public void testFullRebalanceNotTriggeredWhenClientNodeStopsDuringPme() throws Exception {
+ startGrids(2);
+
+ IgniteEx ig0 = grid(0);
+
+ ig0.cluster().active(true);
+
+ IgniteCache<Object, Object> cache = ig0.cache(CACHE_NAME);
+
+ startClientGrid(5);
+
+ final int entryCnt = PARTS_CNT * 1000;
+
+ for (int i = 0; i < entryCnt; i++)
+ cache.put(i, i);
+
+ forceCheckpoint();
+
+ stopGrid(1);
+
+ for (int i = 0; i < entryCnt; i++)
+ cache.put(i, i + 100);
+
+ forceCheckpoint();
+
+ final CountDownLatch exchangeLatch = new CountDownLatch(1);
+
+ final CountDownLatch hangingPmeStartedLatch = new CountDownLatch(1);
+
+ ig0.context().cache().context().exchange().registerExchangeAwareComponent(new PartitionsExchangeAware() {
+ @Override public void onInitAfterTopologyLock(GridDhtPartitionsExchangeFuture fut) {
+ try {
+ hangingPmeStartedLatch.countDown();
+
+ exchangeLatch.await();
+ }
+ catch (InterruptedException e) {
+ e.printStackTrace();
+ }
+ }
+ });
+
+ IgniteInternalFuture fut = GridTestUtils.runAsync(() -> {
+ try {
+ startGrid(1);
+
+ awaitPartitionMapExchange();
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ }
+ });
+
+ IgniteInternalFuture clientStopFut = GridTestUtils.runAsync(() -> {
+ try {
+ hangingPmeStartedLatch.await();
+
+ stopGrid(5);
+ }
+ catch (Exception e) {
+ e.printStackTrace();
+ }
+ }, "client-starter");
+
+ assertFalse(fut.isDone());
+
+ exchangeLatch.countDown();
+
+ clientStopFut.get();
+
+ fut.get();
+
+ awaitPartitionMapExchange();
+
+ boolean histRebalanceInvoked = RebalanceCheckingCommunicationSpi.histRebalances();
+
+ boolean fullRebalanceInvoked = RebalanceCheckingCommunicationSpi.fullRebalances();
+
+ RebalanceCheckingCommunicationSpi.cleanup();
+
+ assertTrue("Historical rebalance hasn't been invoked.", histRebalanceInvoked);
+
+ assertFalse("Full rebalance has been invoked.", fullRebalanceInvoked);
+ }
+
+ /**
+ * Wrapper of communication spi to detect if rebalance has happened.
+ */
+ public static class RebalanceCheckingCommunicationSpi extends TestRecordingCommunicationSpi {
+ /** */
+ private static boolean topVersForHist = false;
+
+ /** */
+ private static boolean topVersForFull = false;
+
+ /** Lock object. */
+ private static final Object mux = new Object();
+
+ /**
+ * @return {@code true} if WAL rebalance has been used.
+ */
+ static boolean histRebalances() {
+ synchronized (mux) {
+ return topVersForHist;
+ }
+ }
+
+ /**
+ * @return {@code true} if full rebalance has been used.
+ */
+ static boolean fullRebalances() {
+ synchronized (mux) {
+ return topVersForFull;
+ }
+ }
+
+ /**
+ * Cleans all rebalances histories.
+ */
+ public static void cleanup() {
+ synchronized (mux) {
+ topVersForHist = false;
+ topVersForFull = false;
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure<IgniteException> ackC) throws IgniteSpiException {
+ if (((GridIoMessage)msg).message() instanceof GridDhtPartitionDemandMessage) {
+ GridDhtPartitionDemandMessage demandMsg = (GridDhtPartitionDemandMessage)((GridIoMessage)msg).message();
+
+ IgniteDhtDemandedPartitionsMap map = demandMsg.partitions();
+
+ if (!map.historicalMap().isEmpty()) {
+ synchronized (mux) {
+ topVersForHist = true;
+ }
+ }
+
+ if (!map.fullSet().isEmpty()) {
+ synchronized (mux) {
+ topVersForFull = true;
+ }
+ }
+ }
+
+ super.sendMessage(node, msg, ackC);
+ }
+ }
+}
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java
index d281e6a..16313f9 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgnitePdsTestSuite2.java
@@ -38,6 +38,7 @@ import org.apache.ignite.internal.processors.cache.persistence.baseline.IgniteAb
import org.apache.ignite.internal.processors.cache.persistence.baseline.IgniteAllBaselineNodesOnlineFullApiSelfTest;
import org.apache.ignite.internal.processors.cache.persistence.baseline.IgniteOfflineBaselineNodeFullApiSelfTest;
import org.apache.ignite.internal.processors.cache.persistence.baseline.IgniteOnlineNodeOutOfBaselineFullApiSelfTest;
+import org.apache.ignite.internal.processors.cache.persistence.db.FullHistRebalanceOnClientStopTest;
import org.apache.ignite.internal.processors.cache.persistence.db.IgnitePdsRebalancingOnNotStableTopologyTest;
import org.apache.ignite.internal.processors.cache.persistence.db.IgnitePdsReserveWalSegmentsTest;
import org.apache.ignite.internal.processors.cache.persistence.db.IgnitePdsReserveWalSegmentsWithCompactionTest;
@@ -182,6 +183,8 @@ public class IgnitePdsTestSuite2 {
GridTestUtils.addTestIfNeeded(suite, IgnitePersistentStoreDataStructuresTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite, FullHistRebalanceOnClientStopTest.class, ignoredTests);
+
// Failover test
GridTestUtils.addTestIfNeeded(suite, IgniteWalFlushFailoverTest.class, ignoredTests);