This is an automated email from the ASF dual-hosted git repository.
zstan 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 cfb6e1f476c IGNITE-27678 Same partitions on different nodes can hold
different updates if writeThrough is enabled (#12925)
cfb6e1f476c is described below
commit cfb6e1f476cff2dfa7db97225c06e49dc703457d
Author: Evgeniy Stanilovskiy <[email protected]>
AuthorDate: Mon Jun 29 09:45:52 2026 +0300
IGNITE-27678 Same partitions on different nodes can hold different updates
if writeThrough is enabled (#12925)
---
.../ignite/internal/CoreMessagesProvider.java | 2 +
.../distributed/dht/GridDhtTxFinishFuture.java | 38 ++-
.../cache/distributed/dht/GridDhtTxLocal.java | 6 +-
.../cache/distributed/dht/GridDhtTxRemote.java | 9 +-
.../distributed/dht/GridDhtTxSalvageMessage.java | 51 ++++
.../preloader/GridDhtPartitionsExchangeFuture.java | 1 +
.../dht/topology/GridDhtLocalPartition.java | 4 +-
.../distributed/near/GridNearTxFinishFuture.java | 9 +
.../cache/store/GridCacheStoreManagerAdapter.java | 13 +
.../cache/transactions/IgniteTxAdapter.java | 2 +-
.../cache/transactions/IgniteTxHandler.java | 14 +
.../cache/transactions/IgniteTxManager.java | 63 +++-
.../transactions/IgniteTxRemoteStateAdapter.java | 2 +-
.../transactions/IgniteTxRemoteStateImpl.java | 4 +-
.../cache/transactions/IgniteTxStateImpl.java | 13 +-
.../IgniteTxCacheWithWriteThroughCheckTest.java | 316 +++++++++++++++++++++
...WriteSynchronizationModesMultithreadedTest.java | 14 +
.../dht/IgniteCacheTxRecoveryRollbackTest.java | 6 +-
.../cache/query/IndexingSpiQueryTxSelfTest.java | 32 ++-
.../ignite/testsuites/IgniteCacheTestSuite19.java | 2 +
20 files changed, 570 insertions(+), 31 deletions(-)
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
index f40b14826cc..1af6f8945cf 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/CoreMessagesProvider.java
@@ -101,6 +101,7 @@ import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxFini
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 class CoreMessagesProvider extends
AbstractMarshallableMessageFactoryProv
withNoSchema(StatisticsRequest.class);
withNoSchema(StatisticsResponse.class);
withNoSchema(CacheContinuousQueryBatchAck.class);
+ withNoSchema(GridDhtTxSalvageMessage.class);
withSchema(CacheContinuousQueryEntry.class);
withNoSchema(QueryInlineSizesDataBagItem.class);
withSchema(QueryProposalsDataBagItem.class);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
index 98c580203c2..a2befc5d999 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxFinishFuture.java
@@ -473,7 +473,7 @@ public final class GridDhtTxFinishFuture<K, V> extends
GridCacheCompoundIdentity
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 @@ public final class GridDhtTxFinishFuture<K, V> extends
GridCacheCompoundIdentity
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 @@ public final class GridDhtTxFinishFuture<K, V> extends
GridCacheCompoundIdentity
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() {
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
index 93ec7a05543..67b32d38f86 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
@@ -460,8 +460,12 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter
implements GridCacheMa
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;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
index e60041e34ed..c2ad54b41b9 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
+++
b/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 class GridDhtTxRemote extends
GridDistributedTxRemoteAdapter {
/** {@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} */
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxSalvageMessage.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxSalvageMessage.java
new file mode 100644
index 00000000000..ae2d0715cb8
--- /dev/null
+++
b/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;
+ }
+}
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 5c0dc84875a..e3c7b42dc7f 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
@@ -270,6 +270,7 @@ public class GridDhtPartitionsExchangeFuture extends
GridDhtTopologyFutureAdapte
private final Map<ClusterNode, GridDhtPartitionsFullMessage> fullMsgs =
new ConcurrentHashMap<>();
/** */
+ @SuppressWarnings("unused")
@GridToStringInclude
private volatile IgniteInternalFuture<?> partReleaseFut;
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
index 1c565177cbd..34ab379e525 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
@@ -1194,7 +1194,9 @@ public class GridDhtLocalPartition extends
GridCacheConcurrentMapImpl implements
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;
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
index 307bdad29e4..6e8c24979a9 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxFinishFuture.java
+++
b/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.GridCacheVersionedFuture;
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 @@ public final class GridNearTxFinishFuture<K, V> extends
GridCacheCompoundIdentit
});
GridDhtTxFinishRequest req =
checkCommittedRequest(mini.futureId(), true);
+ GridDhtTxSalvageMessage salvageReq = null;
for (UUID backupId : backups) {
ClusterNode backup =
cctx.discovery().node(backupId);
@@ -996,6 +998,13 @@ public final class GridNearTxFinishFuture<K, V> extends
GridCacheCompoundIdentit
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);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
index b175df09a09..115fc73f4c4 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/store/GridCacheStoreManagerAdapter.java
@@ -964,6 +964,11 @@ public abstract class GridCacheStoreManagerAdapter extends
GridCacheManagerAdapt
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 @@ public abstract class GridCacheStoreManagerAdapter extends
GridCacheManagerAdapt
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);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
index 193eb2a4e8a..1529d53b5ce 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
@@ -628,7 +628,7 @@ public abstract class IgniteTxAdapter extends
GridMetadataAwareAdapter implement
/**
* @return Finalization status.
*/
- @Override @Nullable public FinalizationStatus finalizationStatus() {
+ @Override public FinalizationStatus finalizationStatus() {
return finalizing;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
index 392f37d32b0..40dae253a43 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
@@ -59,6 +59,7 @@ import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrep
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.GridDhtTxRemote;
+import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxSalvageMessage;
import
org.apache.ignite.internal.processors.cache.distributed.dht.PartitionUpdateCountersMessage;
import
org.apache.ignite.internal.processors.cache.distributed.dht.TransactionAttributesAwareRequest;
import
org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
@@ -223,6 +224,9 @@ public class IgniteTxHandler {
ctx.io().addCacheHandler(GridDhtTxFinishRequest.class, (UUID nodeId,
GridCacheMessage msg) ->
processDhtTxFinishRequest(nodeId, (GridDhtTxFinishRequest)msg));
+ ctx.io().addCacheHandler(GridDhtTxSalvageMessage.class, (UUID nodeId,
GridCacheMessage msg) ->
+ processDhtTxSalvageRequest((GridDhtTxSalvageMessage)msg));
+
ctx.io().addCacheHandler(GridDhtTxOnePhaseCommitAckRequest.class,
(UUID nodeId, GridCacheMessage msg) ->
processDhtTxOnePhaseCommitAckRequest(nodeId,
(GridDhtTxOnePhaseCommitAckRequest)msg));
@@ -1341,6 +1345,16 @@ public class IgniteTxHandler {
}
}
+ /**
+ * @param req Request.
+ */
+ private void processDhtTxSalvageRequest(GridDhtTxSalvageMessage req) {
+ for (IgniteInternalTx active : ctx.tm().activeTransactions()) {
+ if (active.nearXidVersion().equals(req.version()) && active
instanceof GridDhtTxRemote)
+ ctx.tm().salvageTx(active);
+ }
+ }
+
/**
* @param nodeId Node ID.
* @param req Request.
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index cd2704daf90..7f9298c352e 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -44,6 +44,7 @@ import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteSystemProperties;
import org.apache.ignite.binary.BinaryObjectException;
import org.apache.ignite.cluster.ClusterNode;
+import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.TransactionConfiguration;
import org.apache.ignite.events.DiscoveryEvent;
import org.apache.ignite.failure.FailureContext;
@@ -80,6 +81,7 @@ import
org.apache.ignite.internal.processors.cache.distributed.GridDistributedTx
import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxLocal;
import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxOnePhaseCommitAckRequest;
import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxRemote;
+import
org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxSalvageMessage;
import
org.apache.ignite.internal.processors.cache.distributed.dht.TransactionAttributesAwareRequest;
import
org.apache.ignite.internal.processors.cache.distributed.dht.colocated.GridDhtColocatedLockFuture;
import
org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtInvalidPartitionException;
@@ -130,6 +132,7 @@ import static
org.apache.ignite.IgniteSystemProperties.IGNITE_LONG_OPERATIONS_DU
import static
org.apache.ignite.IgniteSystemProperties.IGNITE_MAX_COMPLETED_TX_COUNT;
import static
org.apache.ignite.IgniteSystemProperties.IGNITE_SLOW_TX_WARN_TIMEOUT;
import static
org.apache.ignite.IgniteSystemProperties.IGNITE_TX_DEADLOCK_DETECTION_MAX_ITERS;
+import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC;
import static
org.apache.ignite.configuration.TransactionConfiguration.TX_AWARE_QUERIES_SUPPORTED_MODES;
import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
import static org.apache.ignite.events.EventType.EVT_NODE_JOINED;
@@ -141,6 +144,7 @@ import static
org.apache.ignite.internal.processors.cache.GridCacheOperation.REA
import static
org.apache.ignite.internal.processors.cache.GridCacheUtils.isNearEnabled;
import static
org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx.FinalizationStatus.RECOVERY_FINISH;
import static
org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx.FinalizationStatus.USER_FINISH;
+import static
org.apache.ignite.internal.processors.cache.transactions.IgniteTxStateImpl.deriveSyncMode;
import static
org.apache.ignite.internal.processors.security.SecurityUtils.securitySubjectId;
import static org.apache.ignite.internal.util.GridConcurrentFactory.newMap;
import static org.apache.ignite.transactions.TransactionState.ACTIVE;
@@ -3123,8 +3127,28 @@ public class IgniteTxManager extends
GridCacheSharedManagerAdapter {
", failedNodeId=" + evtNodeId + ']');
for (final IgniteInternalTx tx : activeTransactions()) {
- if ((tx.near() && !tx.local() &&
tx.originatingNodeId().equals(evtNodeId))
- || (tx.storeWriteThrough() &&
tx.masterNodeIds().contains(evtNodeId))) {
+ Map<UUID, Collection<UUID>> txNodes =
tx.transactionNodes();
+
+ if (tx.storeWriteThrough() && txNodes != null
+ && tx.near() && txNodes.containsKey(evtNodeId)
+ && (tx.state() == PREPARING || tx.state() == PREPARED
+ || tx.state() == COMMITTING || tx.state() ==
COMMITTED)) {
+ // Send a message, tx is applied on near node and can
be processed on backup if postponed.
+ sendTxSalvage(tx, evtNodeId);
+ }
+
+ if (tx.storeWriteThrough() &&
tx.masterNodeIds().contains(evtNodeId) && tx.eventNodeId().equals(evtNodeId))
+ salvageTx(tx, RECOVERY_FINISH);
+ else if (tx.storeWriteThrough() &&
!tx.masterNodeIds().contains(cctx.localNodeId())
+ && tx.nodeId().equals(evtNodeId) && tx.state() ==
PREPARED) {
+ // Delay a commit, on backup. It will be raised
further after near or coord. node will confirm it.
+ // In different from {@code FULL_SYNC} modes,
appropriate recovery message can never be raized.
+ if (deriveSyncMode(cctx, tx.txState().cacheIds()) !=
FULL_SYNC)
+ cctx.time().schedule(() -> salvageTx(tx,
RECOVERY_FINISH), 1000, -1);
+ }
+ else if ((tx.near() && !tx.local() &&
tx.originatingNodeId().equals(evtNodeId))
+ || (tx.storeWriteThrough() &&
tx.masterNodeIds().contains(evtNodeId)) &&
+ !tx.eventNodeId().equals(cctx.localNodeId())) {
// Invalidate transactions.
salvageTx(tx, RECOVERY_FINISH);
}
@@ -3166,6 +3190,41 @@ public class IgniteTxManager extends
GridCacheSharedManagerAdapter {
}
}
+ /**
+ * Salvage progress can be postponed in special case when {@link
CacheConfiguration#setWriteThrough} is enabled
+ * and current node is backup. It required for eliminate situaltions
when backup node read data before primary will
+ * continue with transaction commit.
+ *
+ * @see CacheConfiguration#setWriteThrough
+ */
+ private void sendTxSalvage(IgniteInternalTx tx, UUID evtNodeId) {
+ Collection<UUID> involvedNodes =
tx.transactionNodes().get(evtNodeId);
+
+ if (involvedNodes != null) {
+ GridDhtTxSalvageMessage salvageReq = null;
+
+ for (UUID nodeId : involvedNodes) {
+ if (tx.masterNodeIds().contains(nodeId))
+ continue;
+
+ ClusterNode backupNode = cctx.discovery().node(nodeId);
+
+ if (backupNode != null && !backupNode.isLocal()) {
+ if (salvageReq == null)
+ salvageReq = new
GridDhtTxSalvageMessage(tx.nearXidVersion());
+
+ try {
+ cctx.io().send(nodeId, salvageReq, tx.ioPolicy());
+ }
+ catch (IgniteCheckedException e) {
+ log.warning("Failed to send salvage message
[failedNodeId=" + evtNodeId +
+ ", nodeId=" + nodeId + ']', e);
+ }
+ }
+ }
+ }
+ }
+
/**
* @param tx Tx.
* @param failedNode Failed node.
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateAdapter.java
index 60248017ee1..d1826f7cc5a 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateAdapter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateAdapter.java
@@ -33,7 +33,7 @@ import static
org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_ASYNC;
*/
public abstract class IgniteTxRemoteStateAdapter implements
IgniteTxRemoteState {
/** Active cache IDs. */
- private GridIntList activeCacheIds = new GridIntList();
+ private final GridIntList activeCacheIds = new GridIntList();
/** {@inheritDoc} */
@Override public boolean implicitSingle() {
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateImpl.java
index 409cffab11a..7556efe6faf 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxRemoteStateImpl.java
@@ -40,11 +40,11 @@ import org.apache.ignite.internal.util.typedef.internal.S;
public class IgniteTxRemoteStateImpl extends IgniteTxRemoteStateAdapter {
/** Read set. */
@GridToStringInclude
- protected Map<IgniteTxKey, IgniteTxEntry> readMap;
+ protected final Map<IgniteTxKey, IgniteTxEntry> readMap;
/** Write map. */
@GridToStringInclude
- protected Map<IgniteTxKey, IgniteTxEntry> writeMap;
+ protected final Map<IgniteTxKey, IgniteTxEntry> writeMap;
/**
* @param readMap Read map.
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxStateImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxStateImpl.java
index 3577c114add..0ae820a82e0 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxStateImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxStateImpl.java
@@ -167,10 +167,19 @@ public class IgniteTxStateImpl extends
IgniteTxLocalStateAdapter {
/** {@inheritDoc} */
@Override public CacheWriteSynchronizationMode
syncMode(GridCacheSharedContext cctx) {
+ return deriveSyncMode(cctx, activeCacheIds);
+ }
+
+ /**
+ * @param cctx Context.
+ * @param cacheIds Processed caches.
+ * @return Write synchronization mode.
+ */
+ public static CacheWriteSynchronizationMode
deriveSyncMode(GridCacheSharedContext<?, ?> cctx, GridIntList cacheIds) {
CacheWriteSynchronizationMode syncMode =
CacheWriteSynchronizationMode.FULL_ASYNC;
- for (int i = 0; i < activeCacheIds.size(); i++) {
- int cacheId = activeCacheIds.get(i);
+ for (int i = 0; i < cacheIds.size(); i++) {
+ int cacheId = cacheIds.get(i);
CacheWriteSynchronizationMode cacheSyncMode =
cctx.cacheContext(cacheId).config().getWriteSynchronizationMode();
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxCacheWithWriteThroughCheckTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxCacheWithWriteThroughCheckTest.java
new file mode 100644
index 00000000000..b58264ff600
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxCacheWithWriteThroughCheckTest.java
@@ -0,0 +1,316 @@
+/*
+ * 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.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
+import javax.cache.Cache;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.cache.store.CacheStoreAdapter;
+import org.apache.ignite.cluster.ClusterState;
+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.events.DiscoveryEvent;
+import org.apache.ignite.internal.GridTopic;
+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.GridMessageListener;
+import org.apache.ignite.internal.managers.discovery.DiscoCache;
+import org.apache.ignite.internal.managers.eventstorage.DiscoveryEventListener;
+import org.apache.ignite.internal.managers.eventstorage.HighPriorityListener;
+import
org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
+import
org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
+import
org.apache.ignite.internal.processors.cache.transactions.IgniteTxManager;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lang.IgniteBiInClosure;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.ignite.events.EventType.EVT_NODE_FAILED;
+import static org.apache.ignite.events.EventType.EVT_NODE_LEFT;
+import static org.apache.ignite.testframework.GridTestUtils.cartesianProduct;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+
+/** */
+@RunWith(Parameterized.class)
+public class IgniteTxCacheWithWriteThroughCheckTest extends
GridCommonAbstractTest {
+ /** Node kill trigger. */
+ private static CountDownLatch nodeKillLatch;
+
+ /** Node left on backup. */
+ private static CountDownLatch nodeLeftRegisteredOnBackup;
+
+ /** */
+ @Parameterized.Parameter(0)
+ public Boolean withPersistence;
+
+ /** */
+ @Parameterized.Parameter(1)
+ public TransactionConcurrency conc;
+
+ /** */
+ @Parameterized.Parameters(name = "withPersistence={0}, concMode={1}")
+ public static Collection<Object[]> parameters() {
+ return cartesianProduct(
+ List.of(true, false),
+ List.of(OPTIMISTIC, PESSIMISTIC)
+ );
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ stopAllGrids();
+
+ if (withPersistence)
+ cleanPersistenceDir();
+
+ nodeKillLatch = new CountDownLatch(1);
+ nodeLeftRegisteredOnBackup = new CountDownLatch(1);
+
+ MapCacheStore.salvagedLatch = new CountDownLatch(1);
+ MapCacheStore.txCoordStoreLatch = new CountDownLatch(2);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ try {
+ for (Ignite node : G.allGrids()) {
+ Collection<IgniteInternalTx> txs =
((IgniteEx)node).context().cache().context().tm().activeTransactions();
+
+ assertTrue("Unfinished txs [node=" + node.name() + ", txs=" +
txs + ']', txs.isEmpty());
+ }
+ }
+ finally {
+ stopAllGrids();
+
+ super.afterTest();
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setDataStorageConfiguration(new DataStorageConfiguration()
+ .setDefaultDataRegionConfiguration(new
DataRegionConfiguration().setPersistenceEnabled(withPersistence)))
+ .setConsistentId(igniteInstanceName)
+ .setCommunicationSpi(new TestRecordingCommunicationSpi());
+ }
+
+ /** Test scenario:
+ * <ul>
+ * <li>Start 3 nodes [node0, node1, node2].</li>
+ * <li>Initialize put operation into transactional cache where [node1]
holds primary partition for such insertion.</li>
+ * <li>Kill [node1] right after tx PREPARE stage is completed (it
triggers tx recovery procedure).</li>
+ * </ul>
+ *
+ * @see IgniteTxManager#salvageTx(IgniteInternalTx)
+ */
+ @Test
+ public void testTxCoordinatorLeftClusterWithEnabledReadWriteThrough()
throws Exception {
+ // Sequential start is important here.
+ IgniteEx nodeCoord = startGrid(0);
+ // Near node.
+ IgniteEx nodePrimary = startGrid(1);
+ // Backup node.
+ IgniteEx nodeBackup = startGrid(2);
+
+ int firstVal = 1;
+ int secondVal = 2;
+
+ nodeCoord.cluster().state(ClusterState.ACTIVE);
+
+ CacheConfiguration<Object, Object> ccfgWithWriteThrough =
configureCache(DEFAULT_CACHE_NAME);
+ IgniteCache<Object, Object> cache =
nodeCoord.createCache(ccfgWithWriteThrough);
+
+ Integer primaryKey = primaryKey(nodePrimary.cache(DEFAULT_CACHE_NAME));
+
+ try (Transaction tx = nodeCoord.transactions().txStart()) {
+ cache.put(primaryKey, firstVal);
+
+ tx.commit();
+ }
+
+ nodeCoord.cluster().state(ClusterState.INACTIVE);
+
+ GridMessageListener lsnr = new GridMessageListener() {
+ @Override public void onMessage(UUID nodeId, Object msg, byte plc)
{
+ if (msg instanceof GridNearTxPrepareResponse) {
+ IgniteTxManager txMgr =
nodeBackup.context().cache().context().tm();
+ Collection<IgniteInternalTx> txs =
txMgr.activeTransactions();
+
+ assertEquals(1, txs.size());
+ IgniteInternalTx idleTx = txs.iterator().next();
+ assertFalse(idleTx.local());
+
+ Map<GridCacheVersion, IgniteInternalTx> activeTx =
GridTestUtils.getFieldValue(txMgr, "idMap");
+ assertEquals(1, activeTx.size());
+
+ nodeKillLatch.countDown();
+
+ U.awaitQuiet(nodeLeftRegisteredOnBackup);
+
+ // Let`s wait until all discovery events have been
processed on backup node.
+ doSleep(1000);
+
+ MapCacheStore.txCoordStoreLatch.countDown();
+ }
+ }
+ };
+
+ nodeCoord.context().io().removeMessageListener(GridTopic.TOPIC_CACHE);
// Remove old cache listener.
+ nodeCoord.context().io().addMessageListener(GridTopic.TOPIC_CACHE,
lsnr); // Register as first listener.
+ nodeCoord.context().cache().context().io().start0(); // Register cache
listener again.
+
+ nodeCoord.cluster().state(ClusterState.ACTIVE);
+ awaitPartitionMapExchange(true, true, null);
+
+ nodeCoord.context().event().addDiscoveryEventListener(new
BeforeRecoveryListener(), EVT_NODE_FAILED, EVT_NODE_LEFT);
+ nodeBackup.context().event().addDiscoveryEventListener(new
BeforeBackupRecoveryListener(), EVT_NODE_FAILED, EVT_NODE_LEFT);
+
+ IgniteInternalFuture<Object> stopFut = GridTestUtils.runAsync(() -> {
+ nodeKillLatch.await();
+ nodePrimary.close();
+ });
+
+ try (Transaction tx = nodeCoord.transactions().txStart(conc,
READ_COMMITTED)) {
+ cache.put(primaryKey, secondVal);
+
+ tx.commit();
+ }
+ catch (Throwable th) {
+ fail("Unexpected exception: " + th);
+ }
+
+ stopFut.get(getTestTimeout());
+
+ awaitPartitionMapExchange();
+
+ assertEquals(secondVal,
nodeCoord.cache(DEFAULT_CACHE_NAME).get(primaryKey));
+ assertEquals(secondVal,
nodeBackup.cache(DEFAULT_CACHE_NAME).get(primaryKey));
+
+ // Check value after restart.
+ stopAllGrids();
+ startGridsMultiThreaded(3);
+
+ awaitPartitionMapExchange(true, true, null);
+
+ for (Ignite ignite : G.allGrids())
+ assertEquals(secondVal,
ignite.getOrCreateCache(ccfgWithWriteThrough).get(primaryKey));
+ }
+
+ /** */
+ private CacheConfiguration<Object, Object> configureCache(String
cacheName) {
+ CacheConfiguration<Object, Object> ccfg = new
CacheConfiguration<>(cacheName);
+ ccfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL);
+ ccfg.setCacheMode(CacheMode.REPLICATED);
+
ccfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+
+ ccfg.setReadThrough(true);
+ ccfg.setWriteThrough(true);
+ ccfg.setCacheStoreFactory(MapCacheStore::new);
+
+ return ccfg;
+ }
+
+ /** */
+ public static class MapCacheStore extends CacheStoreAdapter<Object,
Object> {
+ /** Store map. */
+ private static final Map<Object, Object> map = new
ConcurrentHashMap<>();
+
+ /** */
+ static CountDownLatch salvagedLatch;
+
+ /** */
+ static CountDownLatch txCoordStoreLatch;
+
+ /** {@inheritDoc} */
+ @Override public void loadCache(IgniteBiInClosure<Object, Object> clo,
Object... args) {
+ for (Map.Entry<Object, Object> e : map.entrySet())
+ clo.apply(e.getKey(), e.getValue());
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object load(Object key) {
+ Object val = map.get(key);
+
+ salvagedLatch.countDown();
+
+ return val;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void write(Cache.Entry<?, ?> e) {
+ map.put(e.getKey(), e.getValue());
+
+ txCoordStoreLatch.countDown();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void delete(Object key) {
+ map.remove(key);
+ }
+ }
+
+ /** */
+ private static class BeforeRecoveryListener implements
DiscoveryEventListener, HighPriorityListener {
+ /** {@inheritDoc} */
+ @Override public void onEvent(DiscoveryEvent evt, DiscoCache
discoCache) {
+ U.awaitQuiet(MapCacheStore.txCoordStoreLatch);
+ }
+
+ /** {@inheritDoc} */
+ @Override public int order() {
+ return -1;
+ }
+ }
+
+ /** */
+ private static class BeforeBackupRecoveryListener implements
DiscoveryEventListener, HighPriorityListener {
+ /** {@inheritDoc} */
+ @Override public void onEvent(DiscoveryEvent evt, DiscoCache
discoCache) {
+ nodeLeftRegisteredOnBackup.countDown();
+ }
+
+ /** {@inheritDoc} */
+ @Override public int order() {
+ return -1;
+ }
+ }
+}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxCacheWriteSynchronizationModesMultithreadedTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxCacheWriteSynchronizationModesMultithreadedTest.java
index 7d05a9bde65..a97158ada72 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxCacheWriteSynchronizationModesMultithreadedTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/IgniteTxCacheWriteSynchronizationModesMultithreadedTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.processors.cache.distributed;
+import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Objects;
@@ -37,7 +38,9 @@ import org.apache.ignite.cache.store.CacheStore;
import org.apache.ignite.cache.store.CacheStoreAdapter;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
+import
org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
import org.apache.ignite.internal.util.lang.GridAbsPredicate;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.G;
@@ -93,6 +96,17 @@ public class
IgniteTxCacheWriteSynchronizationModesMultithreadedTest extends Gri
assertTrue(grid(SRVS + i).configuration().isClientMode());
}
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ for (Ignite node : G.allGrids()) {
+ Collection<IgniteInternalTx> txs =
((IgniteEx)node).context().cache().context().tm().activeTransactions();
+
+ assertTrue("Unfinished txs [node=" + node.name() + ", txs=" + txs
+ ']', txs.isEmpty());
+ }
+
+ super.afterTest();
+ }
+
/**
* @throws Exception If failed.
*/
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCacheTxRecoveryRollbackTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCacheTxRecoveryRollbackTest.java
index 937b51cd4de..35f8053d866 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCacheTxRecoveryRollbackTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/dht/IgniteCacheTxRecoveryRollbackTest.java
@@ -36,8 +36,8 @@ import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.IgniteInternalFuture;
-import org.apache.ignite.internal.IgniteKernal;
import org.apache.ignite.internal.TestRecordingCommunicationSpi;
import
org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxFinishRequest;
import
org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareResponse;
@@ -86,7 +86,7 @@ public class IgniteCacheTxRecoveryRollbackTest extends
GridCommonAbstractTest {
@Override protected void afterTest() throws Exception {
try {
for (Ignite node : G.allGrids()) {
- Collection<IgniteInternalTx> txs =
((IgniteKernal)node).context().cache().context().tm().activeTransactions();
+ Collection<IgniteInternalTx> txs =
((IgniteEx)node).context().cache().context().tm().activeTransactions();
assertTrue("Unfinished txs [node=" + node.name() + ", txs=" +
txs + ']', txs.isEmpty());
}
@@ -379,7 +379,7 @@ public class IgniteCacheTxRecoveryRollbackTest extends
GridCommonAbstractTest {
final IgniteCache<Integer, Integer> clientCache =
client.cache(DEFAULT_CACHE_NAME);
IgniteInternalFuture<?> fut = GridTestUtils.runAsync(new
Callable<Void>() {
- @Override public Void call() throws Exception {
+ @Override public Void call() {
log.info("Start put");
clientCache.put(key, 2);
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/IndexingSpiQueryTxSelfTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/IndexingSpiQueryTxSelfTest.java
index 8b158292d52..2c345f773e1 100644
---
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/IndexingSpiQueryTxSelfTest.java
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/IndexingSpiQueryTxSelfTest.java
@@ -22,13 +22,13 @@ import java.util.Iterator;
import java.util.concurrent.Callable;
import javax.cache.Cache;
import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteException;
import org.apache.ignite.IgniteTransactions;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.internal.IgniteEx;
import org.apache.ignite.internal.processors.cache.GridCacheAbstractSelfTest;
-import
org.apache.ignite.internal.transactions.IgniteTxHeuristicCheckedException;
import org.apache.ignite.spi.IgniteSpiAdapter;
import org.apache.ignite.spi.IgniteSpiException;
import org.apache.ignite.spi.indexing.IndexingQueryFilter;
@@ -36,11 +36,14 @@ import org.apache.ignite.spi.indexing.IndexingSpi;
import org.apache.ignite.testframework.GridTestUtils;
import org.apache.ignite.transactions.Transaction;
import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionHeuristicException;
import org.apache.ignite.transactions.TransactionIsolation;
import org.apache.ignite.transactions.TransactionState;
import org.jetbrains.annotations.Nullable;
import org.junit.Test;
+import static org.apache.ignite.testframework.GridTestUtils.waitForCondition;
+
/**
* Indexing Spi transactional query test
*/
@@ -75,7 +78,7 @@ public class IndexingSpiQueryTxSelfTest extends
GridCacheAbstractSelfTest {
assertNotNull(client.cache(DEFAULT_CACHE_NAME));
- doTestIndexingSpiWithTx(client, 0);
+ doTestIndexingSpiWithTx(client, 0, false);
}
/** */
@@ -83,7 +86,7 @@ public class IndexingSpiQueryTxSelfTest extends
GridCacheAbstractSelfTest {
public void testIndexingSpiWithTxLocal() throws Exception {
IgniteEx ignite = (IgniteEx)primaryNode(0, DEFAULT_CACHE_NAME);
- doTestIndexingSpiWithTx(ignite, 0);
+ doTestIndexingSpiWithTx(ignite, 0, true);
}
/** */
@@ -91,13 +94,13 @@ public class IndexingSpiQueryTxSelfTest extends
GridCacheAbstractSelfTest {
public void testIndexingSpiWithTxNotLocal() throws Exception {
IgniteEx ignite = (IgniteEx)primaryNode(0, DEFAULT_CACHE_NAME);
- doTestIndexingSpiWithTx(ignite, 1);
+ doTestIndexingSpiWithTx(ignite, 1, false);
}
/**
* @throws Exception If failed.
*/
- private void doTestIndexingSpiWithTx(IgniteEx ignite, int key) throws
Exception {
+ private void doTestIndexingSpiWithTx(IgniteEx ignite, int key, boolean
heuristic) throws Exception {
final IgniteCache<Integer, Integer> cache =
ignite.cache(DEFAULT_CACHE_NAME);
final IgniteTransactions txs = ignite.transactions();
@@ -107,7 +110,7 @@ public class IndexingSpiQueryTxSelfTest extends
GridCacheAbstractSelfTest {
System.out.println("Run in transaction: " + concurrency + " "
+ isolation);
GridTestUtils.assertThrowsWithCause(new Callable<Void>() {
- @Override public Void call() throws Exception {
+ @Override public Void call() {
Transaction tx;
try (Transaction tx0 = tx = txs.txStart(concurrency,
isolation)) {
@@ -120,9 +123,20 @@ public class IndexingSpiQueryTxSelfTest extends
GridCacheAbstractSelfTest {
return null;
}
- }, IgniteTxHeuristicCheckedException.class);
-
- checkFutures();
+ }, heuristic ? TransactionHeuristicException.class :
IgniteException.class);
+
+ // Cause asynced salvage in action
+ assertTrue(
+ waitForCondition(() -> {
+ try {
+ checkFutures();
+ return true;
+ }
+ catch (AssertionError err) {
+ return false;
+ }
+ }, 5_000)
+ );
}
}
}
diff --git
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite19.java
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite19.java
index 686ef4aa7b2..317f933c7c5 100644
---
a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite19.java
+++
b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite19.java
@@ -24,6 +24,7 @@ import
org.apache.ignite.cache.ClientCreateCacheGroupOnJoinNodeMapsTest;
import
org.apache.ignite.internal.processors.cache.CacheStoreTxPutAllMultiNodeTest;
import
org.apache.ignite.internal.processors.cache.GridCacheOrderedPreloadingSelfTest;
import
org.apache.ignite.internal.processors.cache.IgniteCachePutKeyAttachedBinaryObjectTest;
+import
org.apache.ignite.internal.processors.cache.distributed.IgniteTxCacheWithWriteThroughCheckTest;
import
org.apache.ignite.internal.processors.cache.distributed.rebalancing.GridCacheRabalancingDelayedPartitionMapExchangeSelfTest;
import
org.apache.ignite.internal.processors.cache.distributed.rebalancing.GridCacheRebalanceOrderTest;
import
org.apache.ignite.internal.processors.cache.distributed.rebalancing.GridCacheRebalancingAsyncSelfTest;
@@ -74,6 +75,7 @@ public class IgniteCacheTestSuite19 {
GridTestUtils.addTestIfNeeded(suite,
CleanupRestoredCachesSlowTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
ClientCreateCacheGroupOnJoinNodeMapsTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
IgniteCachePutKeyAttachedBinaryObjectTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite,
IgniteTxCacheWithWriteThroughCheckTest.class, ignoredTests);
return suite;
}
