entry : msg.partitions().entrySet()) {
Integer grpId = entry.getKey();
@@ -1699,7 +1744,7 @@ else if (!grp.isLocal())
entry.getValue(),
null,
msg.partsToReload(cctx.localNodeId(), grpId),
- msg.partitionSizes(grpId),
+ partsSizes.getOrDefault(grpId, Collections.emptyMap()),
msg.topologyVersion());
}
}
@@ -2434,7 +2479,7 @@ public void exchangerBlockingSectionEnd() {
}
/** */
- private boolean currentThreadIsExchanger() {
+ public boolean currentThreadIsExchanger() {
return exchWorker != null && Thread.currentThread() == exchWorker.runner();
}
@@ -3342,4 +3387,124 @@ private AffinityReadyFuture(AffinityTopologyVersion topVer) {
return S.toString(AffinityReadyFuture.class, this, super.toString());
}
}
+
+ /**
+ * That wrapper class allows avoiding the propagation of the user's exceptions into the Exchange thread.
+ */
+ private class PartitionsExchangeAwareWrapper implements PartitionsExchangeAware {
+ /** */
+ private final PartitionsExchangeAware delegate;
+
+ /**
+ * Creates a new wrapper.
+ * @param delegate Delegate.
+ */
+ public PartitionsExchangeAwareWrapper(PartitionsExchangeAware delegate) {
+ this.delegate = delegate;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onInitBeforeTopologyLock(GridDhtPartitionsExchangeFuture fut) {
+ try {
+ delegate.onInitBeforeTopologyLock(fut);
+ }
+ catch (Exception e) {
+ U.warn(log, "Failed to execute exchange callback.", e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onInitAfterTopologyLock(GridDhtPartitionsExchangeFuture fut) {
+ try {
+ delegate.onInitAfterTopologyLock(fut);
+ }
+ catch (Exception e) {
+ U.warn(log, "Failed to execute exchange callback.", e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onDoneBeforeTopologyUnlock(GridDhtPartitionsExchangeFuture fut) {
+ try {
+ delegate.onDoneBeforeTopologyUnlock(fut);
+ }
+ catch (Exception e) {
+ U.warn(log, "Failed to execute exchange callback.", e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public void onDoneAfterTopologyUnlock(GridDhtPartitionsExchangeFuture fut) {
+ try {
+ delegate.onDoneAfterTopologyUnlock(fut);
+ }
+ catch (Exception e) {
+ U.warn(log, "Failed to execute exchange callback.", e);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public int hashCode() {
+ return delegate.hashCode();
+ }
+
+ /** {@inheritDoc} */
+ @SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
+ @Override public boolean equals(Object obj) {
+ return delegate.equals(obj);
+ }
+ }
+
+ /**
+ * Class to limit action count for unique objects.
+ *
+ * NO guarantees of thread safety are provided.
+ */
+ private static class ActionLimiter {
+ /** */
+ private final int limit;
+
+ /**
+ * Internal storage of objects and counters for each of object.
+ */
+ private final Map actionsCnt = new HashMap<>();
+
+ /**
+ * Set of active objects.
+ */
+ private final Set activeObjects = new HashSet<>();
+
+ /**
+ * @param limit Limit.
+ */
+ private ActionLimiter(int limit) {
+ this.limit = limit;
+ }
+
+ /**
+ * Shows if action is allowed for the given object. Adds this object to internal set of active
+ * objects that are still in use.
+ *
+ * @param obj object.
+ */
+ boolean allowAction(T obj) {
+ activeObjects.add(obj);
+
+ int cnt = actionsCnt.computeIfAbsent(obj, o -> new AtomicInteger(0))
+ .incrementAndGet();
+
+ return (cnt <= limit);
+ }
+
+ /**
+ * Removes old objects from limiter's internal storage. All objects that are contained in internal
+ * storage but not in set of active objects, are assumed as 'old'. This method is to be called
+ * after processing of collection of objects to purge limiter's internal storage.
+ */
+ void trim() {
+ actionsCnt.keySet().removeIf(key -> !activeObjects.contains(key));
+
+ activeObjects.clear();
+ }
+ }
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxFinishSync.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxFinishSync.java
index 1f688f64cc9e2..514e78e898086 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxFinishSync.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheTxFinishSync.java
@@ -29,7 +29,6 @@
import org.apache.ignite.internal.processors.cache.version.GridCacheVersion;
import org.apache.ignite.internal.util.future.GridFinishedFuture;
import org.apache.ignite.internal.util.future.GridFutureAdapter;
-import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.lang.IgniteFuture;
import org.jetbrains.annotations.Nullable;
@@ -66,9 +65,16 @@ public void onFinishSend(UUID nodeId, long threadId) {
ThreadFinishSync threadSync = threadMap.get(threadId);
if (threadSync == null)
- threadSync = F.addIfAbsent(threadMap, threadId, new ThreadFinishSync(threadId));
+ threadMap.put(threadId, threadSync = new ThreadFinishSync(threadId));
- threadSync.onSend(nodeId);
+ synchronized (threadSync) {
+ //thread has to create new ThreadFinishSync
+ //if other thread executing onAckReceived method removed previous threadSync object
+ if (threadMap.get(threadId) == null)
+ threadMap.put(threadId, threadSync = new ThreadFinishSync(threadId));
+
+ threadSync.onSend(nodeId);
+ }
}
/**
@@ -104,8 +110,14 @@ public void onDisconnected(IgniteFuture reconnectFut) {
public void onAckReceived(UUID nodeId, long threadId) {
ThreadFinishSync threadSync = threadMap.get(threadId);
- if (threadSync != null)
+ if (threadSync != null) {
threadSync.onReceive(nodeId);
+
+ synchronized (threadSync) {
+ if (threadSync.isEmpty())
+ threadMap.remove(threadId);
+ }
+ }
}
/**
@@ -114,8 +126,14 @@ public void onAckReceived(UUID nodeId, long threadId) {
* @param nodeId Left node ID.
*/
public void onNodeLeft(UUID nodeId) {
- for (ThreadFinishSync threadSync : threadMap.values())
+ for (ThreadFinishSync threadSync : threadMap.values()) {
threadSync.onNodeLeft(nodeId);
+
+ synchronized (threadSync) {
+ if (threadSync.isEmpty())
+ threadMap.remove(threadSync);
+ }
+ }
}
/**
@@ -193,7 +211,7 @@ public void onDisconnected(IgniteFuture reconnectFut) {
* @param nodeId Node ID response received from.
*/
public void onReceive(UUID nodeId) {
- TxFinishSync sync = nodeMap.get(nodeId);
+ TxFinishSync sync = nodeMap.remove(nodeId);
if (sync != null)
sync.onReceive();
@@ -208,6 +226,13 @@ public void onNodeLeft(UUID nodeId) {
if (sync != null)
sync.onNodeLeft();
}
+
+ /**
+ *
+ */
+ private boolean isEmpty() {
+ return nodeMap.isEmpty();
+ }
}
/**
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
index 0db9b4e79a4df..ad83cbe967fb4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
@@ -126,6 +126,10 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
/** {@code True} if tx should skip adding itself to completed version map on finish. */
private boolean skipCompletedVers;
+ /** Transaction label. */
+ @GridToStringInclude
+ @Nullable private String txLbl;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
@@ -147,6 +151,7 @@ public GridDistributedTxRemoteAdapter() {
* @param txSize Expected transaction size.
* @param subjId Subject ID.
* @param taskNameHash Task name hash code.
+ * @param txLbl Transaction label.
*/
public GridDistributedTxRemoteAdapter(
GridCacheSharedContext ctx,
@@ -161,7 +166,8 @@ public GridDistributedTxRemoteAdapter(
long timeout,
int txSize,
@Nullable UUID subjId,
- int taskNameHash
+ int taskNameHash,
+ String txLbl
) {
super(
ctx,
@@ -179,6 +185,7 @@ public GridDistributedTxRemoteAdapter(
taskNameHash);
this.invalidate = invalidate;
+ this.txLbl = txLbl;
commitVersion(commitVer);
@@ -996,6 +1003,11 @@ protected void addExplicit(IgniteTxEntry e) {
}
}
+ /** {@inheritDoc} */
+ @Override public String label() {
+ return txLbl;
+ }
+
/** {@inheritDoc} */
@Override public String toString() {
return GridToStringBuilder.toString(GridDistributedTxRemoteAdapter.class, this, "super", super.toString());
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
index c24d48f9029c6..a5303fc5dad37 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
@@ -788,6 +788,7 @@ else if (log.isDebugEnabled())
* @param taskName Task name.
* @param expiry Expiry policy.
* @param skipVals Skip values flag.
+ * @param txLbl Transaction label.
* @param mvccSnapshot MVCC snapshot.
* @return Get future.
*/
@@ -800,6 +801,7 @@ IgniteInternalFuture> getDhtAllAsync(
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
boolean recovery,
+ @Nullable String txLbl,
MvccSnapshot mvccSnapshot
) {
return getAllAsync0(keys,
@@ -814,6 +816,7 @@ IgniteInternalFuture> getDhtAllAsync(
/*keep cache objects*/true,
recovery,
/*need version*/true,
+ txLbl,
mvccSnapshot);
}
@@ -828,6 +831,7 @@ IgniteInternalFuture> getDhtAllAsync(
* @param taskNameHash Task name hash code.
* @param expiry Expiry policy.
* @param skipVals Skip values flag.
+ * @param txLbl Transaction label.
* @param mvccSnapshot MVCC snapshot.
* @return DHT future.
*/
@@ -842,6 +846,7 @@ public GridDhtFuture> getDhtAsync(UUID reader,
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
boolean recovery,
+ @Nullable String txLbl,
MvccSnapshot mvccSnapshot
) {
GridDhtGetFuture fut = new GridDhtGetFuture<>(ctx,
@@ -856,6 +861,7 @@ public GridDhtFuture> getDhtAsync(UUID reader,
skipVals,
recovery,
addReaders,
+ txLbl,
mvccSnapshot);
fut.init();
@@ -874,6 +880,7 @@ public GridDhtFuture> getDhtAsync(UUID reader,
* @param taskNameHash Task name hash.
* @param expiry Expiry.
* @param skipVals Skip vals flag.
+ * @param txLbl Transaction label.
* @param mvccSnapshot Mvcc snapshot.
* @return Future for the operation.
*/
@@ -889,6 +896,7 @@ GridDhtGetSingleFuture getDhtSingleAsync(
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
boolean recovery,
+ String txLbl,
MvccSnapshot mvccSnapshot
) {
GridDhtGetSingleFuture fut = new GridDhtGetSingleFuture<>(
@@ -904,6 +912,7 @@ GridDhtGetSingleFuture getDhtSingleAsync(
expiry,
skipVals,
recovery,
+ txLbl,
mvccSnapshot);
fut.init();
@@ -933,6 +942,7 @@ protected void processNearSingleGetRequest(final UUID nodeId, final GridNearSing
expiryPlc,
req.skipValues(),
req.recovery(),
+ req.txLabel(),
req.mvccSnapshot());
fut.listen(new CI1>() {
@@ -1038,6 +1048,7 @@ protected void processNearGetRequest(final UUID nodeId, final GridNearGetRequest
expiryPlc,
req.skipValues(),
req.recovery(),
+ req.txLabel(),
req.mvccSnapshot());
fut.listen(new CI1>>() {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
index b37a57dc9cd71..a45bae616b0b9 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
@@ -123,6 +123,9 @@ public final class GridDhtGetFuture extends GridCompoundIdentityFuture extends GridCompoundIdentityFuturecollectionsReducer(keys.size()));
@@ -171,6 +176,7 @@ public GridDhtGetFuture(
this.skipVals = skipVals;
this.recovery = recovery;
this.addReaders = addReaders;
+ this.txLbl = txLbl;
this.mvccSnapshot = mvccSnapshot;
futId = IgniteUuid.randomUuid();
@@ -456,6 +462,7 @@ private IgniteInternalFuture> getAsync(
expiryPlc,
skipVals,
recovery,
+ txLbl,
mvccSnapshot);
}
else {
@@ -480,6 +487,7 @@ private IgniteInternalFuture> getAsync(
expiryPlc,
skipVals,
recovery,
+ txLbl,
mvccSnapshot);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
index 955c8836edbd2..402df697f4db3 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
@@ -112,6 +112,9 @@ public final class GridDhtGetSingleFuture extends GridFutureAdapter extends GridFutureAdapter entries) {
skipStore,
cctx.store().configured(),
keepBinary,
- cctx.deploymentEnabled());
+ cctx.deploymentEnabled(),
+ inTx() ? tx.label() : null);
try {
for (ListIterator it = dhtMapping.listIterator(); it.hasNext(); ) {
@@ -1322,8 +1323,8 @@ void onResult(GridDhtLockResponse res) {
replicate ? DR_PRELOAD : DR_NONE,
false)) {
if (rec && !entry.isInternal())
- cctx.events().addEvent(entry.partition(), entry.key(), cctx.localNodeId(),
- (IgniteUuid)null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, info.value(), true, null,
+ cctx.events().addEvent(entry.partition(), entry.key(), cctx.localNodeId(), null,
+ null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, info.value(), true, null,
false, null, null, null, false);
}
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
index 76c3b163d577a..95786be854a85 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
@@ -92,6 +92,9 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
/** TTL for read operation. */
private long accessTtl;
+ /** Transaction label. */
+ private String txLbl;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
@@ -123,6 +126,7 @@ public GridDhtLockRequest() {
* @param storeUsed Cache store used flag.
* @param keepBinary Keep binary flag.
* @param addDepInfo Deployment info flag.
+ * @param txLbl Transaction label.
*/
public GridDhtLockRequest(
int cacheId,
@@ -147,7 +151,8 @@ public GridDhtLockRequest(
boolean skipStore,
boolean storeUsed,
boolean keepBinary,
- boolean addDepInfo
+ boolean addDepInfo,
+ String txLbl
) {
super(cacheId,
nodeId,
@@ -179,6 +184,8 @@ public GridDhtLockRequest(
this.subjId = subjId;
this.taskNameHash = taskNameHash;
this.accessTtl = accessTtl;
+
+ this.txLbl = txLbl;
}
/**
@@ -309,6 +316,13 @@ public long accessTtl() {
return accessTtl;
}
+ /**
+ * @return Transaction label.
+ */
+ @Nullable public String txLabel() {
+ return txLbl;
+ }
+
/** {@inheritDoc} */
@Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
super.prepareMarshal(ctx);
@@ -423,6 +437,12 @@ public long accessTtl() {
writer.incrementState();
+ case 31:
+ if (!writer.writeString("txLbl", txLbl))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -519,6 +539,14 @@ public long accessTtl() {
reader.incrementState();
+ case 31:
+ txLbl = reader.readString("txLbl");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridDhtLockRequest.class);
@@ -531,7 +559,7 @@ public long accessTtl() {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 31;
+ return 32;
}
/** {@inheritDoc} */
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index 295eabf371c27..f5b03d6d85a03 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
@@ -339,7 +339,8 @@ protected GridDhtTransactionalCacheAdapter(GridCacheContext ctx, GridCache
req.txSize(),
req.subjectId(),
req.taskNameHash(),
- !req.skipStore() && req.storeUsed());
+ !req.skipStore() && req.storeUsed(),
+ req.txLabel());
tx = ctx.tm().onCreated(null, tx);
@@ -1122,6 +1123,7 @@ public IgniteInternalFuture lockAllAsync(
null,
req.subjectId(),
req.taskNameHash(),
+ req.txLabel(),
null);
if (req.syncCommit())
@@ -2159,6 +2161,7 @@ public GridDhtTxLocal initTxTopologyVersion(UUID nodeId,
null,
txSubjectId,
txTaskNameHash,
+ null,
null);
// if (req.syncCommit())
@@ -2266,7 +2269,8 @@ private void processDhtTxQueryEnlistRequest(UUID primary, GridDhtTxQueryEnlistRe
-1,
req0.subjectId(),
req0.taskNameHash(),
- false);
+ false,
+ null);
tx.mvccSnapshot(new MvccSnapshotWithoutTxs(req0.coordinatorVersion(), req0.counter(),
MVCC_OP_COUNTER_NA, req0.cleanupVersion()));
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 ec92c402a288b..42c9dbe36ce7c 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
@@ -98,6 +98,9 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
@GridToStringExclude
private volatile GridDhtTxPrepareFuture prepFut;
+ /** Transaction label. */
+ private @Nullable String lb;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
@@ -120,6 +123,7 @@ public GridDhtTxLocal() {
* @param storeEnabled Store enabled flag.
* @param txSize Expected transaction size.
* @param txNodes Transaction nodes mapping.
+ * @param lb Transaction label.
* @param parentTx Transaction from which this transaction was copied by(if it was).
*/
public GridDhtTxLocal(
@@ -145,6 +149,7 @@ public GridDhtTxLocal(
Map> txNodes,
UUID subjId,
int taskNameHash,
+ @Nullable String lb,
GridNearTxLocal parentTx
) {
super(
@@ -165,6 +170,8 @@ public GridDhtTxLocal(
subjId,
taskNameHash);
+ this.lb = lb;
+
assert nearNodeId != null;
assert nearFutId != null;
assert nearXidVer != null;
@@ -538,6 +545,11 @@ public IgniteInternalFuture commitDhtLocalAsync() {
return commitDhtLocalAsync();
}
+ /** {@inheritDoc} */
+ @Nullable @Override public String label() {
+ return lb;
+ }
+
/** {@inheritDoc} */
@Override protected void clearPrepareFuture(GridDhtTxPrepareFuture fut) {
assert optimistic();
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
index a7816c5eebdba..d1a096d67eaaf 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
@@ -1926,8 +1926,8 @@ void onResult(GridDhtTxPrepareResponse res) {
drType,
false)) {
if (rec && !entry.isInternal())
- cacheCtx.events().addEvent(entry.partition(), entry.key(), cctx.localNodeId(),
- (IgniteUuid)null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, info.value(), true, null,
+ cacheCtx.events().addEvent(entry.partition(), entry.key(), cctx.localNodeId(), null,
+ null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, info.value(), true, null,
false, null, null, null, false);
if (retVal && !invoke)
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
index 46a06d781b576..5b69f2bd7806a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
@@ -114,6 +114,10 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
/** {@code True} if remote tx should skip adding itself to completed versions map on finish. */
private boolean skipCompletedVers;
+ /** Transaction label. */
+ @GridToStringInclude
+ @Nullable private String txLbl;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
@@ -177,6 +181,7 @@ public GridDhtTxPrepareRequest(
this.nearXidVer = nearXidVer;
this.subjId = subjId;
this.taskNameHash = taskNameHash;
+ this.txLbl = tx.label();
this.mvccSnapshot = mvccSnapshot;
this.updCntrs = updCntrs;
@@ -332,6 +337,13 @@ public boolean skipCompletedVersion() {
return skipCompletedVers;
}
+ /**
+ * @return Transaction label.
+ */
+ @Nullable public String txLabel() {
+ return txLbl;
+ }
+
/**
* {@inheritDoc}
*
@@ -508,6 +520,12 @@ public boolean skipCompletedVersion() {
writer.incrementState();
case 35:
+ if (!writer.writeString("txLbl", txLbl))
+ return false;
+
+ writer.incrementState();
+
+ case 36:
if (!writer.writeCollection("updCntrs", updCntrs, MessageCollectionItemType.MSG))
return false;
@@ -642,6 +660,14 @@ public boolean skipCompletedVersion() {
reader.incrementState();
case 35:
+ txLbl = reader.readString("txLbl");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 36:
updCntrs = reader.readCollection("updCntrs", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
@@ -661,7 +687,7 @@ public boolean skipCompletedVersion() {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 36;
+ return 37;
}
/** {@inheritDoc} */
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 259c3119e8772..9518a014b5029 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
@@ -94,6 +94,7 @@ public GridDhtTxRemote() {
* @param nearXidVer Near transaction ID.
* @param txNodes Transaction nodes mapping.
* @param storeWriteThrough Cache store write through flag.
+ * @param txLbl Transaction label.
*/
public GridDhtTxRemote(
GridCacheSharedContext ctx,
@@ -115,7 +116,8 @@ public GridDhtTxRemote(
@Nullable UUID subjId,
int taskNameHash,
boolean single,
- boolean storeWriteThrough) {
+ boolean storeWriteThrough,
+ @Nullable String txLbl) {
super(
ctx,
nodeId,
@@ -129,7 +131,8 @@ public GridDhtTxRemote(
timeout,
txSize,
subjId,
- taskNameHash
+ taskNameHash,
+ txLbl
);
assert nearNodeId != null;
@@ -169,6 +172,7 @@ public GridDhtTxRemote(
* @param timeout Timeout.
* @param txSize Expected transaction size.
* @param storeWriteThrough Cache store write through flag.
+ * @param txLbl Transaction label.
*/
public GridDhtTxRemote(
GridCacheSharedContext ctx,
@@ -188,7 +192,8 @@ public GridDhtTxRemote(
int txSize,
@Nullable UUID subjId,
int taskNameHash,
- boolean storeWriteThrough) {
+ boolean storeWriteThrough,
+ @Nullable String txLbl) {
super(
ctx,
nodeId,
@@ -202,7 +207,8 @@ public GridDhtTxRemote(
timeout,
txSize,
subjId,
- taskNameHash
+ taskNameHash,
+ txLbl
);
assert nearNodeId != null;
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
index 52658f0045945..22053f38ef77d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
@@ -62,6 +62,8 @@
*/
public class GridPartitionedGetFuture extends CacheDistributedGetFutureAdapter
implements MvccSnapshotResponseListener {
+ /** Transaction label. */
+ private String txLbl;
/** */
protected final MvccSnapshot mvccSnapshot;
@@ -83,6 +85,7 @@ public class GridPartitionedGetFuture extends CacheDistributedGetFutureAda
* @param skipVals Skip values flag.
* @param needVer If {@code true} returns values as tuples containing value and version.
* @param keepCacheObjects Keep cache objects flag.
+ * @param txLbl Transaction label.
* @param mvccSnapshot Mvcc snapshot.
*/
public GridPartitionedGetFuture(
@@ -98,6 +101,7 @@ public GridPartitionedGetFuture(
boolean skipVals,
boolean needVer,
boolean keepCacheObjects,
+ @Nullable String txLbl,
@Nullable MvccSnapshot mvccSnapshot
) {
super(
@@ -119,6 +123,8 @@ public GridPartitionedGetFuture(
this.mvccSnapshot = mvccSnapshot;
+ this.txLbl = txLbl;
+
initLogger(GridPartitionedGetFuture.class);
}
@@ -285,6 +291,7 @@ private void initialMap(AffinityTopologyVersion topVer) {
expiryPlc,
skipVals,
recovery,
+ txLbl,
mvccSnapshot()
);
@@ -491,6 +498,7 @@ private boolean localGet(AffinityTopologyVersion topVer, KeyCacheObject key, int
if (evt) {
cctx.events().readEvent(key,
null,
+ txLbl,
row.value(),
subjId,
taskName,
@@ -697,6 +705,7 @@ public MiniFuture(
skipVals,
cctx.deploymentEnabled(),
recovery,
+ txLbl,
mvccSnapshot()
);
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
index 2314615f58b65..9924635fb6efe 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
@@ -31,6 +31,7 @@
import org.apache.ignite.internal.IgniteDiagnosticAware;
import org.apache.ignite.internal.IgniteDiagnosticPrepareContext;
import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.NodeStoppingException;
import org.apache.ignite.internal.cluster.ClusterTopologyCheckedException;
import org.apache.ignite.internal.cluster.ClusterTopologyServerNotFoundException;
import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion;
@@ -152,6 +153,9 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter getAsync0(KeyCacheObject key,
needVer,
false,
recovery,
+ null,
null);
fut.init();
@@ -1498,6 +1499,7 @@ private IgniteInternalFuture> getAllAsync0(@Nullable Collection loadAsync(
@@ -408,6 +411,7 @@ public final IgniteInternalFuture