IGNITE-27678 Same partitions on different nodes can hold different updates if writeThrough is enabled (#12925)

Author: zstan

modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java

@@ -101,6 +101,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxOnePhaseCommitAckRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareResponse;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxSalvageMessage;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtUnlockRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
 import org.apache.ignite.internal.processors.cache.distributed.dht.TransactionAttributesAwareRequest;
@@ -585,6 +586,7 @@ public CoreMessagesProvider(Marshaller dfltMarsh, Marshaller schemaAwareMarsh, C
         withNoSchema(StatisticsRequest.class);
         withNoSchema(StatisticsResponse.class);
         withNoSchema(CacheContinuousQueryBatchAck.class);
+        withNoSchema(GridDhtTxSalvageMessage.class);
         withSchema(CacheContinuousQueryEntry.class);
         withNoSchema(QueryInlineSizesDataBagItem.class);
         withSchema(QueryProposalsDataBagItem.class);

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

@@ -473,7 +473,7 @@ private boolean finish(boolean commit,
                 if (isNull(cctx.discovery().getAlive(n.id()))) {
                     log.error("Unable to send message (node left topology): " + n);
 
-                    fut.onNodeLeft();
+                    fut.onNodeLeft(n.id());
                 }
                 else {
                     cctx.tm().sendTransactionMessage(n, req, tx, tx.ioPolicy());
@@ -493,7 +493,7 @@ private boolean finish(boolean commit,
             catch (IgniteCheckedException e) {
                 // Fail the whole thing.
                 if (e instanceof ClusterTopologyCheckedException)
-                    fut.onNodeLeft();
+                    fut.onNodeLeft(n.id());
                 else {
                     if (msgLog.isDebugEnabled()) {
                         msgLog.debug("DHT finish fut, failed to send request dht [txId=" + tx.nearXidVersion() +
@@ -687,6 +687,40 @@ void onResult(Throwable e) {
             onDone(e);
         }
 
+        /** */
+        private void onNodeLeft(UUID nodeId) {
+            // Cause in common case #onNodeLeft() completes with no error it`s necessary to send salvage message.
+            if (tx.storeWriteThrough()) {
+                Map<UUID, Collection<UUID>> txNodes = tx.transactionNodes();
+
+                if (txNodes != null) {
+                    Collection<UUID> backups = txNodes.get(nodeId);
+
+                    if (!F.isEmpty(backups)) {
+                        GridDhtTxSalvageMessage salvageReq = null;
+
+                        for (UUID backupId : backups) {
+                            ClusterNode backup = cctx.discovery().node(backupId);
+
+                            if (backup != null && !backup.isLocal()) {
+                                if (salvageReq == null)
+                                    salvageReq = new GridDhtTxSalvageMessage(tx.nearXidVersion());
+
+                                try {
+                                    cctx.io().send(backup, salvageReq, tx.ioPolicy());
+                                }
+                                catch (IgniteCheckedException e) {
+                                    U.error(log, "Failed to send " + GridDhtTxSalvageMessage.class.getName() + " message.", e);
+                                }
+                            }
+                        }
+                    }
+                }
+            }
+
+            onNodeLeft();
+        }
+
         /**
          */
         void onNodeLeft() {

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

@@ -460,8 +460,12 @@ else if (!lockFut.isDone()) {
             logTxFinishErrorSafe(log, commit, e);
 
             // Treat heuristic exception as critical.
-            if (X.hasCause(e, IgniteTxHeuristicCheckedException.class))
+            if (X.hasCause(e, IgniteTxHeuristicCheckedException.class)) {
+                if (storeWriteThrough() && local())
+                    salvageTx();
+
                 cctx.kernalContext().failure().process(new FailureContext(FailureType.CRITICAL_ERROR, e));
+            }
 
             err = e;
         }

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

@@ -17,9 +17,9 @@
 
 package org.apache.ignite.internal.processors.cache.distributed.dht;
 
-import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import javax.cache.processor.EntryProcessor;
@@ -226,12 +226,7 @@ public GridDhtTxRemote(
 
     /** {@inheritDoc} */
     @Override public Collection<UUID> masterNodeIds() {
-        Collection<UUID> res = new ArrayList<>(2);
-
-        res.add(nearNodeId);
-        res.add(nodeId);
-
-        return res;
+        return List.of(nearNodeId, nodeId);
     }
 
     /** {@inheritDoc} */

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

@@ -0,0 +1,51 @@
+/*
+ * 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;
+
+import org.apache.ignite.internal.Order;
+import org.apache.ignite.internal.processors.cache.GridCacheMessage;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+
+/** Salvage tx. */
+public class GridDhtTxSalvageMessage extends GridCacheMessage {
+    /** */
+    @Order(0)
+    GridCacheVersion ver;
+
+    /** Empty constructor. */
+    public GridDhtTxSalvageMessage() {
+        // No-op.
+    }
+
+    /**
+     * @param ver Global transaction identifier within cluster, assigned by transaction coordinator.
+     */
+    public GridDhtTxSalvageMessage(GridCacheVersion ver) {
+        this.ver = ver;
+    }
+
+    /** Tx version. */
+    public GridCacheVersion version() {
+        return ver;
+    }
+
+    /** {@inheritDoc} */
+    @Override public boolean addDeploymentInfo() {
+        return addDepInfo;
+    }
+}

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

@@ -270,6 +270,7 @@ public class GridDhtPartitionsExchangeFuture extends GridDhtTopologyFutureAdapte
     private final Map<ClusterNode, GridDhtPartitionsFullMessage> fullMsgs = new ConcurrentHashMap<>();
 
     /** */
+    @SuppressWarnings("unused")
     @GridToStringInclude
     private volatile IgniteInternalFuture<?> partReleaseFut;
 

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

@@ -1194,7 +1194,9 @@ public void clearDeferredDeletes() {
         while (true) {
             long state = this.state.get();
 
-            assert getPartState(state) != EVICTED;
+            GridDhtPartitionState partState = getPartState(state);
+
+            assert partState != EVICTED : partState;
 
             if (this.state.compareAndSet(state, setSize(state, getSize(state) - 1)))
                 return;

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java

@@ -40,6 +40,7 @@
 import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxMapping;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishRequest;
 import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFinishResponse;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxSalvageMessage;
 import org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
 import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
 import org.apache.ignite.internal.processors.tracing.MTC;
@@ -983,6 +984,7 @@ ClusterNode primary() {
                             });
 
                             GridDhtTxFinishRequest req = checkCommittedRequest(mini.futureId(), true);
+                            GridDhtTxSalvageMessage salvageReq = null;
 
                             for (UUID backupId : backups) {
                                 ClusterNode backup = cctx.discovery().node(backupId);
@@ -996,6 +998,13 @@ ClusterNode primary() {
                                     else {
                                         try {
                                             cctx.io().send(backup, req, tx.ioPolicy());
+
+                                            if (tx.storeWriteThrough()) {
+                                                if (salvageReq == null)
+                                                    salvageReq = new GridDhtTxSalvageMessage(tx.xidVersion());
+
+                                                cctx.io().send(backup, salvageReq, tx.ioPolicy());
+                                            }
                                         }
                                         catch (ClusterTopologyCheckedException ignored) {
                                             mini.onNodeLeft(backupId, discoThread);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java

@@ -964,6 +964,11 @@ private void notifyCacheStoreSessionListeners(SessionData ses, @Nullable StoreOp
                     lsnr.onSessionStart(locSes);
             }
         }
+        catch (RuntimeException e) {
+            U.error(log, "Exception raised during the notification of cache store session listeners: ", e);
+
+            throw e;
+        }
         catch (Exception e) {
             throw new IgniteCheckedException("Failed to start store session: " + e, e);
         }
@@ -984,6 +989,14 @@ private void sessionEnd0(@Nullable IgniteInternalTx tx, boolean threwEx) throws
                     store.sessionEnd(!threwEx);
             }
         }
+        catch (RuntimeException e) {
+            U.error(log, "Exception raised during the notification of cache store session listeners: ", e);
+
+            if (!threwEx)
+                throw U.cast(e);
+            else
+                throw e;
+        }
         catch (Exception e) {
             if (!threwEx)
                 throw U.cast(e);

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java

@@ -628,7 +628,7 @@ protected void uncommit() {
     /**
      * @return Finalization status.
      */
-    @Override @Nullable public FinalizationStatus finalizationStatus() {
+    @Override public FinalizationStatus finalizationStatus() {
         return finalizing;
     }