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 c63df8c543b75..966eefcde6f6f 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 @@ -104,6 +104,7 @@ import org.apache.ignite.internal.processors.metric.impl.BooleanMetricImpl; import org.apache.ignite.internal.processors.metric.impl.HistogramMetricImpl; import org.apache.ignite.internal.processors.query.schema.SchemaNodeLeaveExchangeWorkerTask; +import org.apache.ignite.internal.processors.rollingupgrade.feature.IgniteFeatureManager; import org.apache.ignite.internal.processors.timeout.GridTimeoutObject; import org.apache.ignite.internal.thread.context.OperationContext; import org.apache.ignite.internal.thread.context.Scope; @@ -161,6 +162,7 @@ import static org.apache.ignite.internal.processors.metric.GridMetricManager.PME_OPS_BLOCKED_DURATION; import static org.apache.ignite.internal.processors.metric.GridMetricManager.PME_OPS_BLOCKED_DURATION_HISTOGRAM; import static org.apache.ignite.internal.processors.metric.GridMetricManager.REBALANCED; +import static org.apache.ignite.internal.processors.rollingupgrade.feature.CoreFeatureRegistry.PART_STATE_FROM_OWNER_FEATURE; import static org.apache.ignite.internal.processors.task.TaskExecutionOptions.options; import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.nodeIds; @@ -524,6 +526,17 @@ public void onCoordinatorInitialized() { * @param cache Cache. */ public void onLocalJoin(DiscoveryEvent evt, DiscoCache cache) { + IgniteFeatureManager features = cctx.kernalContext().rollingUpgrade().features(); + + boolean activationAhead = !cctx.kernalContext().clientNode() + && features.localVersionFeatures().contains(PART_STATE_FROM_OWNER_FEATURE) + && !features.isActive(PART_STATE_FROM_OWNER_FEATURE); + + // On activation every server resends its maps once: an owner that finished its last exchange before the + // activation didn't send its map, and the other nodes no longer write its state. + if (activationAhead) + features.listenActivation(PART_STATE_FROM_OWNER_FEATURE, this::scheduleResendPartitions); + discoLsnr.onEvent(evt, cache); } 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 ef1dd7419e7cd..1f84f4f185efa 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 @@ -142,6 +142,7 @@ import static org.apache.ignite.internal.processors.cache.ExchangeDiscoveryEvents.serverJoinEvent; import static org.apache.ignite.internal.processors.cache.ExchangeDiscoveryEvents.serverLeftEvent; import static org.apache.ignite.internal.processors.cache.persistence.snapshot.IgniteSnapshotManager.isSnapshotOperation; +import static org.apache.ignite.internal.processors.rollingupgrade.feature.CoreFeatureRegistry.PART_STATE_FROM_OWNER_FEATURE; import static org.apache.ignite.internal.util.IgniteUtils.doInParallel; import static org.apache.ignite.internal.util.IgniteUtils.doInParallelUninterruptibly; import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.node2id; @@ -2296,6 +2297,8 @@ private String exchangeTimingsLogMessage(String header, List timings) { boolean cleanIdxRebuildFutures = true; try { + Collection locChangedGrps = new HashSet<>(); + waitUntilNewCachesAreRegistered(); if (err == null && @@ -2349,7 +2352,7 @@ private String exchangeTimingsLogMessage(String header, List timings) { } if (exchCtx.events().hasServerLeft() || activateCluster()) - detectLostPartitions(res); + locChangedGrps.addAll(detectLostPartitions(res)); Map m = U.newHashMap(cctx.cache().cacheGroups().size()); @@ -2391,8 +2394,21 @@ private String exchangeTimingsLogMessage(String header, List timings) { cleanIdxRebuildFutures = false; - for (CacheGroupContext grp : cctx.cache().cacheGroups()) - grp.topology().onExchangeDone(this, grp.affinity().readyAffinity(res), false); + for (CacheGroupContext grp : cctx.cache().cacheGroups()) { + if (grp.topology().onExchangeDone(this, grp.affinity().readyAffinity(res), false)) + locChangedGrps.add(grp); + } + + if (!locChangedGrps.isEmpty() + && cctx.kernalContext().rollingUpgrade().features().isActive(PART_STATE_FROM_OWNER_FEATURE)) { + if (log.isDebugEnabled()) { + log.debug("Refresh partitions due to local partition changes at the end of exchange [grps=" + + locChangedGrps.stream().map(CacheGroupContext::cacheOrGroupName) + .collect(Collectors.toList()) + ']'); + } + + cctx.exchange().refreshPartitions(locChangedGrps); + } if (changedAffinity()) cctx.walState().disableGroupDurabilityForPreloading(this); @@ -3500,8 +3516,11 @@ private void findCounterForReservation( * {@link GridDhtPartitionsFullMessage}. * * @param resTopVer Result topology version. + * @return Cache groups where the local partition map changed. */ - private void detectLostPartitions(AffinityTopologyVersion resTopVer) { + private Collection detectLostPartitions(AffinityTopologyVersion resTopVer) { + Collection changedGrps = ConcurrentHashMap.newKeySet(); + try { // Reserve at least 2 threads for system operations. doInParallelUninterruptibly( @@ -3509,7 +3528,8 @@ private void detectLostPartitions(AffinityTopologyVersion resTopVer) { cctx.kernalContext().pools().getSystemExecutorService(), cctx.affinity().cacheGroups().values(), desc -> { - partitionTopology(desc.groupId()).detectLostPartitions(resTopVer, this); + if (partitionTopology(desc.groupId()).detectLostPartitions(resTopVer, this)) + changedGrps.add(cctx.cache().cacheGroup(desc.groupId())); return null; }); @@ -3519,6 +3539,8 @@ private void detectLostPartitions(AffinityTopologyVersion resTopVer) { } timeBag.finishGlobalStage("Detect lost partitions"); + + return changedGrps; } /** diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java index 750f99f3767c4..a8b381a2f872d 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridClientPartitionTopology.java @@ -67,6 +67,7 @@ import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.LOST; import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.MOVING; import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING; +import static org.apache.ignite.internal.processors.rollingupgrade.feature.CoreFeatureRegistry.PART_STATE_FROM_OWNER_FEATURE; import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.nodeIds; /** @@ -1014,9 +1015,9 @@ else if (isStaleUpdate(cur, parts)) { } /** {@inheritDoc} */ - @Override public void onExchangeDone(GridDhtPartitionsExchangeFuture fut, AffinityAssignment assignment, + @Override public boolean onExchangeDone(GridDhtPartitionsExchangeFuture fut, AffinityAssignment assignment, boolean updateRebalanceVer) { - // No-op. + return false; } /** {@inheritDoc} */ @@ -1041,7 +1042,8 @@ else if (isStaleUpdate(cur, parts)) { // Calculate how loss data is handled. boolean safe = partLossPlc != PartitionLossPolicy.IGNORE || !compatibleWithIgnorePlc; - boolean changed = false; + boolean ownerReports = + cctx.kernalContext().rollingUpgrade().features().isActive(PART_STATE_FROM_OWNER_FEATURE); for (int part = 0; part < parts; part++) { boolean lost = F.contains(lostParts, part); @@ -1071,7 +1073,7 @@ else if (isStaleUpdate(cur, parts)) { } } - if (lost) { + if (lost && !ownerReports) { // Update remote maps according to policy. for (Map.Entry entry : node2part.entrySet()) { if (entry.getValue().get(part) != null) @@ -1080,7 +1082,7 @@ else if (isStaleUpdate(cur, parts)) { } } - return changed; + return false; } finally { lock.writeLock().unlock(); diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java index 0c2f415a7a1c1..8f9ee9c90d214 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java @@ -343,11 +343,13 @@ public boolean update(@Nullable GridDhtPartitionExchangeId exchId, /** * Checks if there is at least one owner for each partition in the cache topology for a local node. * If not marks such a partition as LOST or OWNING depending on a policy. + * Changes only local partitions: the other nodes learn the new states from this node's partition map. Until + * {@code PART_STATE_FROM_OWNER_FEATURE} is active, it also writes them into its copies of the other nodes' maps. * * @param resTopVer Exchange result version. * @param fut Exchange futute for topology events to detect. * - * @return {@code True} if partitions state got updated. + * @return {@code True} if the local partition map changed. */ public boolean detectLostPartitions(AffinityTopologyVersion resTopVer, GridDhtPartitionsExchangeFuture fut); @@ -452,10 +454,13 @@ public Map> resetOwners( /** * Callback on exchange done. * + * @param fut Exchange future, or {@code null} if called outside an exchange. * @param assignment New affinity assignment. * @param updateRebalanceVer {@code True} if need check rebalance state. + * @return {@code True} if the local partition map changed. */ - public void onExchangeDone(GridDhtPartitionsExchangeFuture fut, AffinityAssignment assignment, boolean updateRebalanceVer); + public boolean onExchangeDone(GridDhtPartitionsExchangeFuture fut, AffinityAssignment assignment, + boolean updateRebalanceVer); /** * Rents a partition and updates a partition map if the partition was switched to RENTING. diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java index afbb0c191f682..38d716006c577 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopologyImpl.java @@ -84,6 +84,7 @@ import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.MOVING; import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING; import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.RENTING; +import static org.apache.ignite.internal.processors.rollingupgrade.feature.CoreFeatureRegistry.PART_STATE_FROM_OWNER_FEATURE; import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.nodeIds; /** @@ -1606,6 +1607,8 @@ private boolean shouldOverridePartitionMap(GridDhtPartitionMap currentMap, GridD return false; } + boolean changed = false; + if (exchangeVer != null) { assert exchangeVer.compareTo(readyTopVer) >= 0 && exchangeVer.compareTo(lastTopChangeVer) >= 0; @@ -1616,6 +1619,9 @@ private boolean shouldOverridePartitionMap(GridDhtPartitionMap currentMap, GridD this.lostParts = new HashSet<>(lostParts); if (exchFut != null && !exchFut.activateCluster()) { + boolean ownerReports = + ctx.kernalContext().rollingUpgrade().features().isActive(PART_STATE_FROM_OWNER_FEATURE); + for (Integer part : lostParts) { GridDhtLocalPartition locPart = localPartition(part); @@ -1623,7 +1629,8 @@ private boolean shouldOverridePartitionMap(GridDhtPartitionMap currentMap, GridD // part.clearFuture lifecycle will be broken after resetting. // New partition should be created instead. if (locPart != null && locPart.state() != EVICTED) { - locPart.markLost(); + if (locPart.markLost() && ownerReports) + changed = true; GridDhtPartitionMap locMap = partMap.get(ctx.localNodeId()); @@ -1672,8 +1679,6 @@ private boolean shouldOverridePartitionMap(GridDhtPartitionMap currentMap, GridD diffFromAffinityVer = readyTopVer; } - boolean changed = false; - GridDhtPartitionMap nodeMap = partMap.get(ctx.localNodeId()); // Only in real exchange occurred. @@ -2049,9 +2054,11 @@ else if (isStaleUpdate(cur, parts)) { } /** {@inheritDoc} */ - @Override public void onExchangeDone(@Nullable GridDhtPartitionsExchangeFuture fut, + @Override public boolean onExchangeDone(@Nullable GridDhtPartitionsExchangeFuture fut, AffinityAssignment assignment, boolean updateRebalanceVer) { + boolean changed = false; + lock.writeLock().lock(); try { @@ -2082,19 +2089,27 @@ else if (isStaleUpdate(cur, parts)) { // Own orphan moving partitions (having no suppliers). if (fut != null && (fut.events().hasServerJoin() || fut.changedBaseline())) - ownOrphans(); + changed = ownOrphans(); } finally { lock.writeLock().unlock(); } + + return changed; } /** * Check if some local moving partitions have no suitable supplier and own them. * This can happen if a partition has been created by affinity assignment on new node and no supplier exists. * For example, if a node joins topology being a first data node for cache with node filter configured. + * + * @return {@code True} if a local partition was owned. */ - private void ownOrphans() { + private boolean ownOrphans() { + boolean ownerReports = ctx.kernalContext().rollingUpgrade().features().isActive(PART_STATE_FROM_OWNER_FEATURE); + + boolean changed = false; + for (int p = 0; p < grp.affinity().partitions(); p++) { boolean hasOwner = false; @@ -2110,15 +2125,24 @@ private void ownOrphans() { GridDhtLocalPartition locPart = localPartition(p); if (locPart != null && locPart.state() != EVICTED && locPart.state() != LOST) { - locPart.own(); + if (locPart.own()) { + changed = true; + + if (ownerReports) + updateLocal(p, OWNING, updateSeq.incrementAndGet(), readyTopVer); + } - for (GridDhtPartitionMap map : node2part.values()) { - if (map.get(p) != null) - map.put(p, OWNING); + if (!ownerReports) { + for (GridDhtPartitionMap map : node2part.values()) { + if (map.get(p) != null) + map.put(p, OWNING); + } } } } } + + return changed; } /** @@ -2225,6 +2249,9 @@ private void rebuildDiff(AffinityAssignment affAssignment) { boolean changed = false; + boolean ownerReports = + ctx.kernalContext().rollingUpgrade().features().isActive(PART_STATE_FROM_OWNER_FEATURE); + for (int part = 0; part < grp.affinity().partitions(); part++) { boolean lost = F.contains(lostParts, part); @@ -2260,9 +2287,11 @@ private void rebuildDiff(AffinityAssignment affAssignment) { final GridDhtPartitionState prevState = locPart.state(); - changed = isLossIgnored ? locPart.own() : locPart.markLost(); + boolean locChanged = isLossIgnored ? locPart.own() : locPart.markLost(); + + if (locChanged) { + changed = true; - if (changed) { long updSeq = updateSeq.incrementAndGet(); updateLocal(locPart.id(), locPart.state(), updSeq, resTopVer); @@ -2274,14 +2303,16 @@ private void rebuildDiff(AffinityAssignment affAssignment) { } // Update remote maps according to policy. - for (Map.Entry entry : node2part.entrySet()) { - if (entry.getKey().equals(ctx.localNodeId())) - continue; + if (!ownerReports) { + for (Map.Entry entry : node2part.entrySet()) { + if (entry.getKey().equals(ctx.localNodeId())) + continue; - GridDhtPartitionState p0 = entry.getValue().get(part); + GridDhtPartitionState p0 = entry.getValue().get(part); - if (p0 != null && p0 != EVICTED) - entry.getValue().put(part, isLossIgnored ? OWNING : LOST); + if (p0 != null && p0 != EVICTED) + entry.getValue().put(part, isLossIgnored ? OWNING : LOST); + } } } } diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java index 4d00489327fea..ff7a69d72b6ed 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/CoreFeatureRegistry.java @@ -111,4 +111,10 @@ public class CoreFeatureRegistry { /** Feature that introduces the snapshot list command. */ public static final IgniteFeature SNAPSHOT_LIST_FEATURE = new IgniteCoreFeature(4); + + /** + * Feature that makes only the node holding a partition write its state after loss detection or after the node owns + * a partition that no node has. The other nodes take the state from that node's partition map. + */ + public static final IgniteFeature PART_STATE_FROM_OWNER_FEATURE = new IgniteCoreFeature(5); } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheNoSupplierPartitionMapTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheNoSupplierPartitionMapTest.java new file mode 100644 index 0000000000000..c7ef7962e5169 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CacheNoSupplierPartitionMapTest.java @@ -0,0 +1,150 @@ +/* + * 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; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.UUID; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.TestRecordingCommunicationSpi; +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap; +import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition; +import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState; +import org.apache.ignite.internal.util.typedef.internal.CU; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.apache.ignite.util.AttributeNodeFilter; +import org.junit.Test; + +import static org.apache.ignite.internal.TestRecordingCommunicationSpi.blockDemandMessageForGroup; +import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; + +/** + * Checks that the other nodes see the partitions that a data node owns at the end of an exchange because no node has + * them. Only the data node writes OWNING into its partition map, so it must send the map. + *

+ * The cache keeps its data only on nodes with the {@link #DATA_ATTR} attribute. The coordinator, another server and a + * client have no data and watch the data node's partitions in their partition maps. The data node joins outside the + * baseline and gets its partitions when the baseline changes. + *

+ * A node that joins the baseline also rebalances the system cache, and the affinity change exchange after that + * rebalance carries the node's map anyway. The test holds the system cache's rebalance on the data node, as if the + * system cache were large. + */ +public class CacheNoSupplierPartitionMapTest extends GridCommonAbstractTest { + /** */ + private static final String CACHE = "cache"; + + /** */ + private static final int PARTS = 32; + + /** Attribute of the nodes that keep the cache's data. */ + private static final String DATA_ATTR = "data"; + + /** Index of the client node. */ + private static final int CLIENT = 2; + + /** Index of the data node. */ + private static final int DATA = 3; + + /** Time to wait for the partition states, in milliseconds. */ + private static final long WAIT = 30_000; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + boolean data = igniteInstanceName.equals(getTestIgniteInstanceName(DATA)); + + return super.getConfiguration(igniteInstanceName) + .setUserAttributes(Collections.singletonMap(DATA_ATTR, data)) + .setCommunicationSpi(new TestRecordingCommunicationSpi()) + .setCacheConfiguration(new CacheConfiguration(CACHE) + .setBackups(0) + .setAffinity(new RendezvousAffinityFunction(false, PARTS)) + .setNodeFilter(new AttributeNodeFilter(DATA_ATTR, true))); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + stopAllGrids(); + + super.afterTest(); + } + + /** + * The first data node gets all partitions when it joins the baseline and owns them, since no node has them. + * + * @throws Exception If failed. + */ + @Test + public void testOrphanPartitionsOwned() throws Exception { + IgniteEx crd = startGrid(0); + + crd.cluster().baselineAutoAdjustEnabled(false); + + startGrid(1); + startClientGrid(CLIENT); + + IgniteEx data = startGrid(DATA); + + TestRecordingCommunicationSpi.spi(data) + .blockMessages(blockDemandMessageForGroup(CU.cacheId(CU.UTILITY_CACHE_NAME))); + + crd.cluster().setBaselineTopology(crd.cluster().topologyVersion()); + + assertTrue("The data node didn't own its partitions", waitForCondition(() -> { + List parts = data.cachex(CACHE).context().topology().localPartitions(); + + return parts.size() == PARTS && parts.stream().allMatch(p -> p.state() == OWNING); + }, WAIT)); + + UUID dataId = data.localNode().id(); + + for (int idx = 0; idx <= CLIENT; idx++) { + IgniteEx watcher = grid(idx); + + if (!waitForCondition(() -> notOwning(watcher, dataId).isEmpty(), WAIT)) { + fail("The partition map of " + watcher.name() + " differs from the data node: " + + notOwning(watcher, dataId)); + } + } + } + + /** + * @param watcher Node whose partition map is checked. + * @param dataId ID of the data node. + * @return Partitions that the watcher's copy of the data node's map doesn't show OWNING. + */ + private List notOwning(IgniteEx watcher, UUID dataId) { + GridDhtPartitionMap view = watcher.cachex(CACHE).context().topology().partitionMap(false).get(dataId); + + List res = new ArrayList<>(); + + for (int p = 0; p < PARTS; p++) { + GridDhtPartitionState seen = view == null ? null : view.get(p); + + if (seen != OWNING) + res.add("p=" + p + " seen=" + seen); + } + + return res; + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLossStaleMapTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLossStaleMapTest.java new file mode 100644 index 0000000000000..e1fd0a2229f92 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/CachePartitionLossStaleMapTest.java @@ -0,0 +1,602 @@ +/* + * 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; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.Comparator; +import java.util.List; +import java.util.UUID; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.atomic.AtomicBoolean; +import org.apache.ignite.Ignite; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cache.PartitionLossPolicy; +import org.apache.ignite.cache.affinity.AffinityFunction; +import org.apache.ignite.cache.affinity.AffinityFunctionContext; +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.IgniteConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.NodeStoppingException; +import org.apache.ignite.internal.TestRecordingCommunicationSpi; +import org.apache.ignite.internal.managers.communication.GridIoMessage; +import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; +import org.apache.ignite.internal.processors.cache.IgniteInternalCache; +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionMap; +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsExchangeFuture; +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsFullMessage; +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage; +import org.apache.ignite.internal.processors.cache.distributed.dht.topology.EvictionContext; +import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtLocalPartition; +import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState; +import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopology; +import org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionTopologyImpl; +import org.apache.ignite.internal.processors.resource.DependencyResolver; +import org.apache.ignite.internal.util.typedef.G; +import org.apache.ignite.internal.util.typedef.internal.CU; +import org.apache.ignite.internal.util.typedef.internal.U; +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.TestDependencyResolver; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; + +import static java.util.concurrent.TimeUnit.MILLISECONDS; +import static org.apache.ignite.cache.PartitionLossPolicy.IGNORE; +import static org.apache.ignite.cache.PartitionLossPolicy.READ_WRITE_SAFE; +import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.EVICTED; +import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.MOVING; +import static org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.RENTING; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; + +/** + * Checks that partition maps match the local partition states of the nodes after a node leaves with the only copy of + * some partitions. + *

+ * The new primary recreates each lost partition empty as MOVING when it applies the coordinator's full message. At + * the end of the exchange it owns the partition under the {@link PartitionLossPolicy#IGNORE IGNORE} policy or marks + * it LOST under the safe policies. Only the new primary writes this state into its partition map and sends the map. + * A map the node sent earlier shows the partition MOVING or doesn't have it at all. The other nodes must take the + * new primary's state in the end, whatever order its maps arrive in. + *

+ * In a real cluster such an earlier map is a resend scheduled by {@code scheduleResendPartitions()}, for example after + * an eviction. {@link #testStaleCoordinatorMapAfterEviction()} checks the case where the coordinator is the new + * primary. + */ +public class CachePartitionLossStaleMapTest extends GridCommonAbstractTest { + /** */ + protected static final String CACHE = "cache"; + + /** */ + private static final int PARTS = 64; + + /** Index of the node that sends a stale partition map before it owns a lost partition or marks it LOST. */ + private static final int STALE_MAP_SENDER = 1; + + /** Index of the node that leaves. */ + private static final int LEAVING = 2; + + /** Index of a server node that watches node 1's partitions in its partition map. */ + private static final int OBSERVER = 3; + + /** Index of the client node. */ + private static final int CLIENT = 4; + + /** Time to wait for a step of a test, in milliseconds. */ + private static final long WAIT = 30_000; + + /** + * When set, the new primary sends its partition map right before it owns a MOVING partition of the cache or marks + * it LOST. + */ + private final AtomicBoolean sendBeforeChange = new AtomicBoolean(); + + /** When set, the next eviction of a partition of the cache waits for {@link #evictionResume}. */ + private final AtomicBoolean holdEviction = new AtomicBoolean(); + + /** Counted down when an eviction starts waiting. */ + private final CountDownLatch evictionHeld = new CountDownLatch(1); + + /** Lets the held eviction go on. */ + private final CountDownLatch evictionResume = new CountDownLatch(1); + + /** When set, the coordinator waits for {@link #crdResume} right after it sends the next exchange full message. */ + private final AtomicBoolean stallAfterFullMsg = new AtomicBoolean(); + + /** Counted down when the coordinator starts waiting. */ + private final CountDownLatch crdStalled = new CountDownLatch(1); + + /** Lets the coordinator go on. */ + private final CountDownLatch crdResume = new CountDownLatch(1); + + /** Partition loss policy of the cache. */ + private PartitionLossPolicy plc = IGNORE; + + /** + * Whether the coordinator is the new primary of all lost partitions. Sets {@link JoinOrderAffinityFunction} and a + * {@link StallingCommunicationSpi}. + */ + private boolean crdNewPrimary; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + cfg.setCacheConfiguration(new CacheConfiguration(CACHE) + .setBackups(0) + .setPartitionLossPolicy(plc) + .setAffinity(crdNewPrimary + ? new JoinOrderAffinityFunction() + : new RendezvousAffinityFunction(false, PARTS))); + + cfg.setCommunicationSpi(crdNewPrimary ? new StallingCommunicationSpi() : new TestRecordingCommunicationSpi()); + + // Evictions run in the rebalance pool, one thread by default below 8 cores: a held one would block the rest. + cfg.setRebalanceThreadPoolSize(4); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + evictionResume.countDown(); + crdResume.countDown(); + + stopAllGrids(); + + super.afterTest(); + } + + /** + * Node 1 sends a stale map before it owns a lost partition under the IGNORE policy. + * + * @throws Exception If failed. + */ + @Test + public void testStaleMapBeforeLostPartitionOwned() throws Exception { + checkStaleMapBeforeLostPartitionChanged(IGNORE); + } + + /** + * Node 1 sends a stale map before it marks a lost partition LOST under the READ_WRITE_SAFE policy. + * + * @throws Exception If failed. + */ + @Test + public void testStaleMapBeforeLostPartitionMarkedLost() throws Exception { + checkStaleMapBeforeLostPartitionChanged(READ_WRITE_SAFE); + } + + /** + * Checks that the partition maps of the coordinator, the observer and the client match the local states of the + * nodes after node 1 sends a stale map. Node 2 joins, takes over partitions from the other servers, and leaves with + * the only copies. Right before node 1 owns a lost partition or marks it LOST, it sends its map, which still shows + * the partition MOVING. The coordinator gets this map once its own exchange is finished. + * + * @param plc Partition loss policy. + * @throws Exception If failed. + */ + private void checkStaleMapBeforeLostPartitionChanged(PartitionLossPolicy plc) throws Exception { + this.plc = plc; + + IgniteEx crd = startNodes(); + + awaitPartitionMapExchange(true, true, null); + + UUID crdId = crd.localNode().id(); + + TestRecordingCommunicationSpi sndSpi = TestRecordingCommunicationSpi.spi(grid(STALE_MAP_SENDER)); + + // Hold node 1's maps sent outside an exchange until the coordinator finishes the exchange. + sndSpi.blockMessages((node, msg) -> node.id().equals(crdId) + && msg instanceof GridDhtPartitionsSingleMessage + && ((GridDhtPartitionsSingleMessage)msg).exchangeId() == null); + + sendBeforeChange.set(true); + + long leaveVer = crd.cluster().topologyVersion() + 1; + + stopGrid(LEAVING); + + assertTrue("Node 1 didn't change a lost partition", waitForCondition(() -> !sendBeforeChange.get(), WAIT)); + + assertTrue("Node 1's map wasn't held", sndSpi.waitForBlocked(1, WAIT)); + + assertTrue(waitForCondition(() -> exchangeFinished(crd, leaveVer), WAIT)); + + sndSpi.stopBlock(); + + checkNodes(crd, grid(OBSERVER), grid(CLIENT)); + } + + /** + * Checks that the partition maps of the coordinator, the observer and the client match the local states of the + * nodes after node 1 sends its map before it gets the full message. + *

+ * When node 1's last RENTING partition is evicted, the map changes and node 1 schedules a resend. Node 1 doesn't + * know its lost partitions until it gets the full message, so the resent map doesn't have them. Its version is + * newer than the one the coordinator has, so the coordinator takes it. The test: + *

    + *
  1. holds the eviction of one partition that node 1 gives to the observer;
    2. + *
  2. stops node 2 and holds the coordinator's full message to node 1;
    3. + *
  3. lets the eviction finish once the coordinator has finished the exchange;
    4. + *
  4. lets the full message go once the coordinator has taken node 1's resent map.
    5. + *
+ * + * @throws Exception If failed. + */ + @Test + public void testStaleMapBeforeFullMessage() throws Exception { + holdEviction.set(true); + + IgniteEx crd = startNodes(); + + awaitPartitionMapExchange(); + + assertTrue("Node 1 didn't start an eviction", evictionHeld.await(WAIT, MILLISECONDS)); + + IgniteEx snd = grid(STALE_MAP_SENDER); + + GridDhtPartitionTopology sndTop = snd.cachex(CACHE).context().topology(); + + // Wait until only the held eviction is left: the last eviction schedules the resend. + assertTrue(waitForCondition(() -> sndTop.localPartitions().stream() + .filter(p -> p.state() == RENTING).count() == 1, WAIT)); + + UUID sndId = snd.localNode().id(); + + TestRecordingCommunicationSpi crdSpi = TestRecordingCommunicationSpi.spi(crd); + + crdSpi.blockMessages((node, msg) -> node.id().equals(sndId) + && msg instanceof GridDhtPartitionsFullMessage + && ((GridDhtPartitionsFullMessage)msg).exchangeId() != null); + + long leaveVer = crd.cluster().topologyVersion() + 1; + + stopGrid(LEAVING); + + assertTrue("The coordinator didn't send the full message", crdSpi.waitForBlocked(1, WAIT)); + + assertTrue(waitForCondition(() -> exchangeFinished(crd, leaveVer), WAIT)); + + GridDhtPartitionTopology crdTop = crd.cachex(CACHE).context().topology(); + + long seq = crdTop.partitionMap(false).get(sndId).updateSequence(); + + evictionResume.countDown(); + + assertTrue("The coordinator didn't take node 1's resent map", waitForCondition( + () -> crdTop.partitionMap(false).get(sndId).updateSequence() > seq, WAIT)); + + crdSpi.stopBlock(); + + checkNodes(crd, grid(OBSERVER), grid(CLIENT)); + } + + /** + * Checks that the coordinator sends its partition map again after it owns the lost partitions assigned to it. + *

+ * The coordinator creates these partitions as MOVING, sends the full message, and only then owns them. When its + * last RENTING partition is evicted, the map changes and the coordinator schedules a resend. If the resend fires + * before the coordinator owns the partitions, it carries them as MOVING. Its version is newer than the full + * message's, so node 1 takes it. Node 1 sees the partitions OWNING only once the coordinator sends its map after + * owning them. + *

+ * The test: + *

    + *
  1. holds the eviction of one partition that the coordinator gave to node 1;
    2. + *
  2. stops node 2 and stalls the coordinator right after it sends the full message;
    3. + *
  3. lets the eviction finish once node 1 has finished the exchange;
    4. + *
  4. lets the coordinator go on once node 1 has taken the resent map.
    5. + *
+ * + * @throws Exception If failed. + */ + @Test + public void testStaleCoordinatorMapAfterEviction() throws Exception { + crdNewPrimary = true; + + IgniteEx crd = startGrid(0, partitionHooksResolver()); + + IgniteCache cache = crd.cache(CACHE); + + for (int i = 0; i < PARTS * 4; i++) + cache.put(i, i); + + // Hold the eviction of one partition that node 1 takes over, as if the partition were large. + holdEviction.set(true); + + IgniteEx other = startGrid(1); + + IgniteEx client = startClientGrid(CLIENT); + + awaitPartitionMapExchange(); + + assertTrue("The coordinator didn't start an eviction", evictionHeld.await(WAIT, MILLISECONDS)); + + startGrid(LEAVING); + + awaitPartitionMapExchange(); + + GridDhtPartitionTopology crdTop = crd.cachex(CACHE).context().topology(); + + // Wait until only the held eviction is left: the last eviction schedules the resend, + // and no copy of node 2's partitions may stay on the coordinator. + assertTrue(waitForCondition(() -> crdTop.localPartitions().stream() + .filter(p -> p.state() == RENTING).count() == 1, WAIT)); + + UUID crdId = crd.localNode().id(); + + stallAfterFullMsg.set(true); + + long leaveVer = crd.cluster().topologyVersion() + 1; + + stopGrid(LEAVING); + + assertTrue("The coordinator didn't send the full message", crdStalled.await(WAIT, MILLISECONDS)); + + assertTrue(waitForCondition(() -> exchangeFinished(other, leaveVer), WAIT)); + + GridDhtPartitionTopology otherTop = other.cachex(CACHE).context().topology(); + + long seq = otherTop.partitionMap(false).get(crdId).updateSequence(); + + evictionResume.countDown(); + + assertTrue("Node 1 didn't take the coordinator's resent map", waitForCondition( + () -> otherTop.partitionMap(false).get(crdId).updateSequence() > seq, WAIT)); + + crdResume.countDown(); + + assertTrue(waitForCondition(() -> exchangeFinished(crd, leaveVer), WAIT)); + + checkNodes(other, client); + } + + /** + * Does nothing: scans and key operations don't fail while a node sees no owner. The SQL subclass runs a query. + * + * @param node Node. + */ + protected void runQuery(Ignite node) { + // No-op. + } + + /** + * Starts the coordinator, node 1 with the partition hooks and the client, loads the cache, then starts the + * observer and node 2. + * + * @return Coordinator. + * @throws Exception If failed. + */ + private IgniteEx startNodes() throws Exception { + IgniteEx crd = startGrid(0); + + startGrid(STALE_MAP_SENDER, partitionHooksResolver()); + + startClientGrid(CLIENT); + + IgniteCache cache = crd.cache(CACHE); + + for (int i = 0; i < PARTS * 4; i++) + cache.put(i, i); + + startGrid(OBSERVER); + + startGrid(LEAVING); + + return crd; + } + + /** + * Runs a query on each viewer. Then checks that the partition map of each viewer matches the local states of the + * nodes. + * + * @param viewers Nodes whose partition maps are checked. + * @throws Exception If failed. + */ + private void checkNodes(IgniteEx... viewers) throws Exception { + for (IgniteEx viewer : viewers) + runQuery(viewer); + + for (IgniteEx viewer : viewers) { + if (!waitForCondition(() -> mismatches(viewer, CACHE).isEmpty(), WAIT)) { + fail("The partition map of " + viewer.name() + " differs from the nodes: " + + mismatches(viewer, CACHE)); + } + } + } + + /** + * Creates a resolver that adds two hooks to the partitions of the cache. Once {@link #sendBeforeChange} is set, the + * node sends its partition map right before the next MOVING partition is owned or marked LOST. Once + * {@link #holdEviction} is set, the next eviction waits for {@link #evictionResume}. + * + * @return Dependency resolver with the partition hooks. + */ + private TestDependencyResolver partitionHooksResolver() { + return new TestDependencyResolver(new DependencyResolver() { + @Override public T resolve(T instance) { + if (instance instanceof GridDhtPartitionTopologyImpl) { + GridDhtPartitionTopologyImpl top = (GridDhtPartitionTopologyImpl)instance; + + top.partitionFactory((ctx, grp, id, recovery) -> new GridDhtLocalPartition(ctx, grp, id, recovery) { + /** {@inheritDoc} */ + @Override public boolean own() { + sendIfArmed(); + + return super.own(); + } + + /** {@inheritDoc} */ + @Override public boolean markLost() { + sendIfArmed(); + + return super.markLost(); + } + + /** {@inheritDoc} */ + @Override protected long clearAll(EvictionContext evictionCtx) throws NodeStoppingException { + if (CACHE.equals(grp.cacheOrGroupName()) && state() == RENTING + && holdEviction.compareAndSet(true, false)) { + evictionHeld.countDown(); + + U.awaitQuiet(evictionResume); + } + + return super.clearAll(evictionCtx); + } + + /** Sends the partition map once, if this is a MOVING partition of the cache. */ + private void sendIfArmed() { + if (CACHE.equals(grp.cacheOrGroupName()) && state() == MOVING + && sendBeforeChange.compareAndSet(true, false)) + ctx.exchange().refreshPartitions(Collections.singleton(grp)); + } + }); + } + + return instance; + } + }); + } + + /** + * @param node Node. + * @param topVer Topology version. + * @return Whether the node has finished the exchange for the topology version. + */ + static boolean exchangeFinished(IgniteEx node, long topVer) { + GridDhtPartitionsExchangeFuture fut = node.context().cache().context().exchange().lastFinishedFuture(); + + return fut != null && fut.topologyVersion().topologyVersion() >= topVer; + } + + /** + * Compares the viewer's partition map with the local partition states of the servers. A client that hasn't started + * the cache keeps the map in a client topology. + * + * @param viewer Server or client whose partition map is checked. + * @param cacheName Cache name. + * @return Partitions whose state in the viewer's map differs from the state on the server that holds them. + */ + static List mismatches(IgniteEx viewer, String cacheName) { + IgniteInternalCache cache = viewer.cachex(cacheName); + + GridDhtPartitionTopology viewerTop = cache != null + ? cache.context().topology() + : viewer.context().cache().context().exchange().clientTopologyIfExists(CU.cacheId(cacheName)); + + assertNotNull(viewer.name(), viewerTop); + + List res = new ArrayList<>(); + + for (Ignite g : G.allGrids()) { + if (g.cluster().localNode().isClient()) + continue; + + GridDhtPartitionMap view = viewerTop.partitionMap(false).get(g.cluster().localNode().id()); + + GridDhtPartitionTopology top = ((IgniteEx)g).cachex(cacheName).context().topology(); + + for (int p = 0; p < top.partitions(); p++) { + GridDhtLocalPartition locPart = top.localPartition(p, AffinityTopologyVersion.NONE, false, true); + + GridDhtPartitionState loc = locPart == null ? null : locPart.state(); + GridDhtPartitionState seen = view == null ? null : view.get(p); + + // The map keeps evicted partitions, the node doesn't. + if (seen == EVICTED) + seen = null; + + if (loc != seen) + res.add(g.name() + " p=" + p + " local=" + loc + " seen=" + seen); + } + } + + return res; + } + + /** + * Once {@link #stallAfterFullMsg} is set, holds the coordinator's thread right after it sends an exchange full + * message, before it owns its lost partitions. The thread waits for {@link #crdResume}. + */ + private class StallingCommunicationSpi extends TestRecordingCommunicationSpi { + /** {@inheritDoc} */ + @Override public void sendMessage(ClusterNode node, Message msg, IgniteInClosure ackC) + throws IgniteSpiException { + super.sendMessage(node, msg, ackC); + + if (msg instanceof GridIoMessage + && ((GridIoMessage)msg).message() instanceof GridDhtPartitionsFullMessage + && ((GridDhtPartitionsFullMessage)((GridIoMessage)msg).message()).exchangeId() != null + && stallAfterFullMsg.compareAndSet(true, false)) { + crdStalled.countDown(); + + U.awaitQuiet(crdResume); + } + } + } + + /** + * Assigns partition {@code p} to the server node with index {@code p % 3} in join order, or to the oldest node if + * there is no such node. + */ + private static class JoinOrderAffinityFunction implements AffinityFunction { + /** */ + private static final long serialVersionUID = 0L; + + /** {@inheritDoc} */ + @Override public void reset() { + // No-op. + } + + /** {@inheritDoc} */ + @Override public int partitions() { + return PARTS; + } + + /** {@inheritDoc} */ + @Override public int partition(Object key) { + return U.safeAbs(key.hashCode()) % PARTS; + } + + /** {@inheritDoc} */ + @Override public List> assignPartitions(AffinityFunctionContext ctx) { + List nodes = new ArrayList<>(ctx.currentTopologySnapshot()); + + nodes.sort(Comparator.comparingLong(ClusterNode::order)); + + List> res = new ArrayList<>(PARTS); + + for (int p = 0; p < PARTS; p++) + res.add(Collections.singletonList(nodes.get(p % 3 < nodes.size() ? p % 3 : 0))); + + return res; + } + + /** {@inheritDoc} */ + @Override public void removeNode(UUID nodeId) { + // No-op. + } + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/PartitionStateFromOwnerRollingUpgradeTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/PartitionStateFromOwnerRollingUpgradeTest.java new file mode 100644 index 0000000000000..11a9824717dcc --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/PartitionStateFromOwnerRollingUpgradeTest.java @@ -0,0 +1,189 @@ +/* + * 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; + +import java.util.List; +import java.util.UUID; +import org.apache.ignite.Ignite; +import org.apache.ignite.cache.affinity.rendezvous.RendezvousAffinityFunction; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.internal.IgniteEx; +import org.apache.ignite.internal.TestRecordingCommunicationSpi; +import org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPartitionsSingleMessage; +import org.apache.ignite.internal.processors.rollingupgrade.AbstractRollingUpgradeTest; +import org.apache.ignite.internal.util.typedef.G; +import org.apache.ignite.plugin.extensions.communication.Message; +import org.junit.Test; + +import static org.apache.ignite.cache.PartitionLossPolicy.IGNORE; +import static org.apache.ignite.internal.processors.cache.distributed.CachePartitionLossStaleMapTest.exchangeFinished; +import static org.apache.ignite.internal.processors.cache.distributed.CachePartitionLossStaleMapTest.mismatches; +import static org.apache.ignite.internal.processors.cache.distributed.dht.preloader.GridDhtPreloader.DFLT_PRELOAD_RESEND_TIMEOUT; +import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; + +/** + * Checks that until {@code PART_STATE_FROM_OWNER_FEATURE} is active the nodes handle lost partitions as the previous + * version does: the other nodes write the new owners' state into their partition maps themselves. Once the feature is + * active, they take it from the owners' maps. + *

+ * In both parts a server leaves with the only copies of some partitions under the IGNORE policy, while the maps that + * the other servers send outside an exchange are held, so the coordinator sees only the state it wrote itself. The + * coordinator and the client watch the new owners' partitions. The client hasn't started the cache, so it keeps the + * cache's partition map in a client topology. + */ +public class PartitionStateFromOwnerRollingUpgradeTest extends AbstractRollingUpgradeTest { + /** */ + private static final String CACHE = "cache"; + + /** */ + private static final int PARTS = 64; + + /** Version without {@code PART_STATE_FROM_OWNER_FEATURE}. */ + private static final String OLD_VER = "2.18.0"; + + /** Index of the client node, as started by {@link #startCluster(String)}. */ + private static final int CLIENT = 2; + + /** Index of the server that leaves. */ + private static final int LEAVING = 3; + + /** Time to wait for the partition maps to match, in milliseconds. */ + private static final long WAIT = 30_000; + + /** + * Before the upgrade the coordinator and the client see the new owners right after the exchange, with the owners' + * maps held. Once all nodes are upgraded and the version is finalized, the feature is active, and they see the new + * owners only after the maps arrive. On activation a server resends its maps, and a client doesn't. + * + * @throws Exception If failed. + */ + @Test + public void testOwnersReportStateAfterUpgrade() throws Exception { + startCluster(OLD_VER); + startGrid(LEAVING, OLD_VER); + + grid(0).createCache(new CacheConfiguration(CACHE) + .setBackups(0) + .setPartitionLossPolicy(IGNORE) + .setAffinity(new RendezvousAffinityFunction(false, PARTS))); + + awaitPartitionMapExchange(); + + IgniteEx oldCrd = stopLeavingWithMapsHeld(); + + for (IgniteEx viewer : new IgniteEx[] {oldCrd, grid(CLIENT)}) { + List mismatches = mismatches(viewer, CACHE); + + assertTrue(viewer.name() + " didn't write the new owners' state: " + mismatches, mismatches.isEmpty()); + } + + for (Ignite g : G.allGrids()) { + if (g != oldCrd && !g.cluster().localNode().isClient()) { + assertFalse(g.name() + " sent its partition map before the feature is active", + TestRecordingCommunicationSpi.spi(g).hasBlockedMessages()); + } + } + + TestRecordingCommunicationSpi.stopBlockAll(); + + ru(0).enableVersionUpgrade(); + + upgradeNodeVersion(1, OLD_VER, MASTER_VER_REF); + upgradeNodeVersion(CLIENT, OLD_VER, MASTER_VER_REF); + upgradeNodeVersion(0, OLD_VER, MASTER_VER_REF); + + awaitPartitionMapExchange(); + + TestRecordingCommunicationSpi srvSpi = TestRecordingCommunicationSpi.spi(grid(coordinatorIndex() == 0 ? 1 : 0)); + TestRecordingCommunicationSpi clientSpi = TestRecordingCommunicationSpi.spi(grid(CLIENT)); + + srvSpi.record((node, msg) -> sentOutsideExchange(msg)); + clientSpi.record((node, msg) -> sentOutsideExchange(msg)); + + // Also checks that every node has the features of the new version active. + finalizeClusterVersion(1, MASTER_VER_REF); + + assertTrue("The server didn't resend its partition map on activation", + waitForCondition(() -> !srvSpi.recordedMessages(false).isEmpty(), WAIT)); + + assertFalse("The client resent its partition map on activation", + waitForCondition(() -> !clientSpi.recordedMessages(false).isEmpty(), DFLT_PRELOAD_RESEND_TIMEOUT)); + + startGrid(LEAVING, MASTER_VER_REF); + + awaitPartitionMapExchange(); + + IgniteEx crd = stopLeavingWithMapsHeld(); + + for (IgniteEx viewer : new IgniteEx[] {crd, grid(CLIENT)}) { + List mismatches = mismatches(viewer, CACHE); + + assertFalse(viewer.name() + " wrote the new owners' state itself", mismatches.isEmpty()); + + assertTrue(viewer.name() + " differs from the nodes in more than the new owners' partitions: " + mismatches, + mismatches.stream().allMatch(m -> m.contains(" local=OWNING "))); + } + + TestRecordingCommunicationSpi.stopBlockAll(); + + for (IgniteEx viewer : new IgniteEx[] {crd, grid(CLIENT)}) { + if (!waitForCondition(() -> mismatches(viewer, CACHE).isEmpty(), WAIT)) + fail("The partition map of " + viewer.name() + " differs from the nodes: " + mismatches(viewer, CACHE)); + } + } + + /** + * Stops the leaving server while the other servers' partition maps sent outside an exchange are held, and waits + * for every node to finish the exchange. + * + * @return Coordinator. + * @throws Exception If failed. + */ + private IgniteEx stopLeavingWithMapsHeld() throws Exception { + IgniteEx crd = grid(coordinatorIndex()); + + UUID crdId = crd.localNode().id(); + + for (Ignite g : G.allGrids()) { + if (g != crd && !g.cluster().localNode().isClient()) { + TestRecordingCommunicationSpi.spi(g).blockMessages( + (node, msg) -> node.id().equals(crdId) && sentOutsideExchange(msg)); + } + } + + long leaveVer = crd.cluster().topologyVersion() + 1; + + stopGrid(LEAVING); + + for (Ignite g : G.allGrids()) { + assertTrue(g.name() + " didn't finish the exchange", + waitForCondition(() -> exchangeFinished((IgniteEx)g, leaveVer), WAIT)); + } + + return crd; + } + + /** + * @param msg Message. + * @return Whether the message is a partition map sent outside an exchange. + */ + private static boolean sentOutsideExchange(Message msg) { + return msg instanceof GridDhtPartitionsSingleMessage + && ((GridDhtPartitionsSingleMessage)msg).exchangeId() == null; + } +} diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite15.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite15.java index cef8063e41e98..66e500005459a 100644 --- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite15.java +++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite15.java @@ -37,6 +37,8 @@ import org.apache.ignite.internal.processors.cache.IgniteCacheStoreCollectionTest; import org.apache.ignite.internal.processors.cache.distributed.CacheLateAffinityAssignmentNodeJoinValidationTest; import org.apache.ignite.internal.processors.cache.distributed.CacheLateAffinityAssignmentTest; +import org.apache.ignite.internal.processors.cache.distributed.CacheNoSupplierPartitionMapTest; +import org.apache.ignite.internal.processors.cache.distributed.CachePartitionLossStaleMapTest; import org.apache.ignite.internal.processors.cache.distributed.GridExchangeFreeCellularSwitchComplexOperationsTest; import org.apache.ignite.internal.processors.cache.distributed.GridExchangeFreeCellularSwitchIsolationTest; import org.apache.ignite.internal.processors.cache.distributed.GridExchangeFreeCellularSwitchTxContinuationTest; @@ -45,6 +47,7 @@ import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheGroupsPartitionLossPolicySelfTest; import org.apache.ignite.internal.processors.cache.distributed.IgniteCachePartitionLossPolicySelfTest; import org.apache.ignite.internal.processors.cache.distributed.IgniteCacheTxIteratorSelfTest; +import org.apache.ignite.internal.processors.cache.distributed.PartitionStateFromOwnerRollingUpgradeTest; import org.apache.ignite.internal.processors.cache.distributed.rebalancing.CacheManualRebalancingTest; import org.apache.ignite.internal.processors.cache.distributed.rebalancing.RebalanceMetricsTest; import org.apache.ignite.internal.processors.cache.distributed.replicated.IgniteCacheSyncRebalanceModeSelfTest; @@ -98,6 +101,9 @@ public static List> suite(Collection ignoredTests) { GridTestUtils.addTestIfNeeded(suite, ClusterActivationFailureTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, CacheCreateDestroyClusterReadOnlyModeTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, IgniteCachePartitionLossPolicySelfTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, CachePartitionLossStaleMapTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, CacheNoSupplierPartitionMapTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, PartitionStateFromOwnerRollingUpgradeTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, IgniteCacheGroupsPartitionLossPolicySelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, CacheRebalancingSelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, CacheManualRebalancingTest.class, ignoredTests); diff --git a/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IndexingCachePartitionLossStaleMapTest.java b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IndexingCachePartitionLossStaleMapTest.java new file mode 100644 index 0000000000000..cd1d5de77ffa3 --- /dev/null +++ b/modules/indexing/src/test/java/org/apache/ignite/internal/processors/cache/IndexingCachePartitionLossStaleMapTest.java @@ -0,0 +1,56 @@ +/* + * 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; + +import org.apache.ignite.Ignite; +import org.apache.ignite.cache.query.SqlFieldsQuery; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.internal.processors.cache.distributed.CachePartitionLossStaleMapTest; +import org.junit.Ignore; +import org.junit.Test; + +import static java.util.concurrent.TimeUnit.SECONDS; + +/** + * Runs {@link CachePartitionLossStaleMapTest} with a SQL query. A query fails if the node that runs it sees a partition + * without an owner until the query's timeout. + */ +public class IndexingCachePartitionLossStaleMapTest extends CachePartitionLossStaleMapTest { + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + for (CacheConfiguration ccfg : cfg.getCacheConfiguration()) + ccfg.setIndexedTypes(Integer.class, Integer.class); + + return cfg; + } + + /** {@inheritDoc} */ + @Override protected void runQuery(Ignite node) { + node.cache(CACHE).query(new SqlFieldsQuery("select count(*) from Integer").setTimeout(10, SECONDS)).getAll(); + } + + /** {@inheritDoc} */ + @Test + @Ignore("Under READ_WRITE_SAFE a SQL query fails on a lost partition by design") + @Override public void testStaleMapBeforeLostPartitionMarkedLost() throws Exception { + super.testStaleMapBeforeLostPartitionMarkedLost(); + } +} diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java index d1d2e840db5b8..43e15c7c31f74 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java +++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteBinaryCacheQueryTestSuite3.java @@ -62,6 +62,7 @@ import org.apache.ignite.internal.processors.cache.IgniteCacheP2pUnmarshallingQueryErrorTest; import org.apache.ignite.internal.processors.cache.IgniteCheckClusterStateBeforeExecuteQueryTest; import org.apache.ignite.internal.processors.cache.IndexingCachePartitionLossPolicySelfTest; +import org.apache.ignite.internal.processors.cache.IndexingCachePartitionLossStaleMapTest; import org.apache.ignite.internal.processors.cache.QueryEntityCaseMismatchTest; import org.apache.ignite.internal.processors.cache.ReservationsOnDoneAfterTopologyUnlockFailTest; import org.apache.ignite.internal.processors.cache.SqlFieldsQuerySelfTest; @@ -293,6 +294,7 @@ // Partition loss. IndexingCachePartitionLossPolicySelfTest.class, + IndexingCachePartitionLossStaleMapTest.class, // Partitions eviction SqlPartitionEvictionTest.class,