IGNITE-29088 Fix stale lost partition states in the coordinator's partition map

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionsExchangeFuture.java

@@ -2316,6 +2316,8 @@ private String exchangeTimingsLogMessage(String header, List<String> timings) {
 
         boolean cleanIdxRebuildFutures = true;
 
+        Collection<CacheGroupContext> locChangedGrps = new HashSet<>();
+
         try {
             waitUntilNewCachesAreRegistered();
 
@@ -2370,7 +2372,7 @@ private String exchangeTimingsLogMessage(String header, List<String> timings) {
                 }
 
                 if (exchCtx.events().hasServerLeft() || activateCluster())
-                    detectLostPartitions(res);
+                    locChangedGrps.addAll(detectLostPartitions(res));
 
                 Map<Integer, CacheGroupValidation> m = U.newHashMap(cctx.cache().cacheGroups().size());
 
@@ -2412,8 +2414,20 @@ private String exchangeTimingsLogMessage(String header, List<String> 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()) {
+                    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);
@@ -3523,16 +3537,20 @@ private void findCounterForReservation(
      * {@link GridDhtPartitionsFullMessage}.
      *
      * @param resTopVer Result topology version.
+     * @return Cache groups where the state of a local partition changed.
      */
-    private void detectLostPartitions(AffinityTopologyVersion resTopVer) {
+    private Collection<CacheGroupContext> detectLostPartitions(AffinityTopologyVersion resTopVer) {
+        Collection<CacheGroupContext> changedGrps = ConcurrentHashMap.newKeySet();
+
         try {
             // Reserve at least 2 threads for system operations.
             doInParallelUninterruptibly(
                 U.availableThreadCount(cctx.kernalContext(), GridIoPolicy.SYSTEM_POOL, 2),
                 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;
                 });
@@ -3542,6 +3560,8 @@ private void detectLostPartitions(AffinityTopologyVersion resTopVer) {
         }
 
         timeBag.finishGlobalStage("Detect lost partitions");
+
+        return changedGrps;
     }
 
     /**

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.SupportedFeatureRegistry.PART_STATE_FROM_OWNER_FEATURE;
 import static org.apache.ignite.internal.util.lang.ClusterNodeFunc.nodeIds;
 
 /**
@@ -1027,9 +1028,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} */
@@ -1056,6 +1057,9 @@ else if (isStaleUpdate(cur, parts)) {
 
             boolean changed = false;
 
+            boolean updRmtMaps =
+                !cctx.kernalContext().rollingUpgrade().features().isActive(PART_STATE_FROM_OWNER_FEATURE);
+
             for (int part = 0; part < parts; part++) {
                 boolean lost = F.contains(lostParts, part);
 
@@ -1084,7 +1088,7 @@ else if (isStaleUpdate(cur, parts)) {
                     }
                 }
 
-                if (lost) {
+                if (lost && updRmtMaps) {
                     // Update remote maps according to policy.
                     for (Map.Entry<UUID, GridDhtPartitionMap> entry : node2part.entrySet()) {
                         if (entry.getValue().get(part) != null)

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtPartitionTopology.java

@@ -347,7 +347,7 @@ public boolean update(@Nullable GridDhtPartitionExchangeId exchId,
      * @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 state of at least one local partition changed.
      */
     public boolean detectLostPartitions(AffinityTopologyVersion resTopVer, GridDhtPartitionsExchangeFuture fut);
 
@@ -454,8 +454,10 @@ public Map<UUID, Set<Integer>> resetOwners(
      *
      * @param assignment New affinity assignment.
      * @param updateRebalanceVer {@code True} if need check rebalance state.
+     * @return {@code True} if the state of at least one local partition 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.

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.SupportedFeatureRegistry.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;
 
@@ -1623,7 +1626,7 @@ 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();
+                                    changed |= locPart.markLost();
 
                                     GridDhtPartitionMap locMap = partMap.get(ctx.localNodeId());
 
@@ -1672,8 +1675,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 +2050,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 +2085,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 updRmtMaps = !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 +2121,23 @@ private void ownOrphans() {
                 GridDhtLocalPartition locPart = localPartition(p);
 
                 if (locPart != null && locPart.state() != EVICTED && locPart.state() != LOST) {
-                    locPart.own();
+                    if (locPart.state() == MOVING && locPart.own()) {
+                        changed = true;
 
-                    for (GridDhtPartitionMap map : node2part.values()) {
-                        if (map.get(p) != null)
-                            map.put(p, OWNING);
+                        updateLocal(p, OWNING, updateSeq.incrementAndGet(), readyTopVer);
+                    }
+
+                    if (updRmtMaps) {
+                        for (GridDhtPartitionMap map : node2part.values()) {
+                            if (map.get(p) != null)
+                                map.put(p, OWNING);
+                        }
                     }
                 }
             }
         }
+
+        return changed;
     }
 
     /**
@@ -2225,6 +2244,9 @@ private void rebuildDiff(AffinityAssignment affAssignment) {
 
                 boolean changed = false;
 
+                boolean updRmtMaps =
+                    !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 +2282,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 +2298,16 @@ private void rebuildDiff(AffinityAssignment affAssignment) {
                         }
 
                         // Update remote maps according to policy.
-                        for (Map.Entry<UUID, GridDhtPartitionMap> entry : node2part.entrySet()) {
-                            if (entry.getKey().equals(ctx.localNodeId()))
-                                continue;
+                        if (updRmtMaps) {
+                            for (Map.Entry<UUID, GridDhtPartitionMap> 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);
+                            }
                         }
                     }
                 }

modules/core/src/main/java/org/apache/ignite/internal/processors/rollingupgrade/feature/SupportedFeatureRegistry.java

@@ -93,4 +93,10 @@
 public class SupportedFeatureRegistry {
     /** */
     public static final IgniteFeature ROLLING_UPGRADE_FEATURE = new IgniteCoreFeature(0);
+
+    /**
+     * After loss detection, and after a node owns a partition that no node has, only the node that holds the partition
+     * writes its state into its partition map. The other nodes take the state from that map.
+     */
+    public static final IgniteFeature PART_STATE_FROM_OWNER_FEATURE = new IgniteCoreFeature(1);
 }