You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2018/11/15 09:09:04 UTC
[1/2] ignite git commit: IGNITE-9274: Added optional labels to
transactions. This closes #4600.
Repository: ignite
Updated Branches:
refs/heads/master d37f1d709 -> 3c54f3894
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
index f594e2b..2531143 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetRequest.java
@@ -108,6 +108,9 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
/** TTL for read operation. */
private long accessTtl;
+ /** Transaction label. */
+ private @Nullable String txLbl;
+
/** */
private MvccSnapshot mvccSnapshot;
@@ -133,6 +136,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
* @param createTtl New TTL to set after entry is created, -1 to leave unchanged.
* @param accessTtl New TTL to set after entry is accessed, -1 to leave unchanged.
* @param addDepInfo Deployment info.
+ * @param txLbl Transaction label.
* @param mvccSnapshot Mvcc snapshot.
*/
public GridNearGetRequest(
@@ -151,6 +155,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
boolean skipVals,
boolean addDepInfo,
boolean recovery,
+ @Nullable String txLbl,
@Nullable MvccSnapshot mvccSnapshot
) {
assert futId != null;
@@ -180,6 +185,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
this.createTtl = createTtl;
this.accessTtl = accessTtl;
this.addDepInfo = addDepInfo;
+ this.txLbl = txLbl;
this.mvccSnapshot = mvccSnapshot;
if (readThrough)
@@ -297,6 +303,15 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
}
/**
+ * Get transaction label (may be null).
+ *
+ * @return Possible transaction label;
+ */
+ @Nullable public String txLabel() {
+ return txLbl;
+ }
+
+ /**
* @param ctx Cache context.
* @throws IgniteCheckedException If failed.
*/
@@ -397,37 +412,43 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
writer.incrementState();
case 9:
- if (!writer.writeCollection("readersFlags", readersFlags, MessageCollectionItemType.BOOLEAN))
+ if (!writer.writeMessage("mvccSnapshot", mvccSnapshot))
return false;
writer.incrementState();
case 10:
- if (!writer.writeUuid("subjId", subjId))
+ if (!writer.writeCollection("readersFlags", readersFlags, MessageCollectionItemType.BOOLEAN))
return false;
writer.incrementState();
case 11:
- if (!writer.writeInt("taskNameHash", taskNameHash))
+ if (!writer.writeUuid("subjId", subjId))
return false;
writer.incrementState();
case 12:
- if (!writer.writeMessage("topVer", topVer))
+ if (!writer.writeInt("taskNameHash", taskNameHash))
return false;
writer.incrementState();
case 13:
- if (!writer.writeMessage("ver", ver))
+ if (!writer.writeMessage("topVer", topVer))
return false;
writer.incrementState();
case 14:
- if (!writer.writeMessage("mvccSnapshot", mvccSnapshot))
+ if (!writer.writeString("txLbl", txLbl))
+ return false;
+
+ writer.incrementState();
+
+ case 15:
+ if (!writer.writeMessage("ver", ver))
return false;
writer.incrementState();
@@ -497,7 +518,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 9:
- readersFlags = reader.readCollection("readersFlags", MessageCollectionItemType.BOOLEAN);
+ mvccSnapshot = reader.readMessage("mvccSnapshot");
if (!reader.isLastRead())
return false;
@@ -505,7 +526,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 10:
- subjId = reader.readUuid("subjId");
+ readersFlags = reader.readCollection("readersFlags", MessageCollectionItemType.BOOLEAN);
if (!reader.isLastRead())
return false;
@@ -513,7 +534,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 11:
- taskNameHash = reader.readInt("taskNameHash");
+ subjId = reader.readUuid("subjId");
if (!reader.isLastRead())
return false;
@@ -521,7 +542,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 12:
- topVer = reader.readMessage("topVer");
+ taskNameHash = reader.readInt("taskNameHash");
if (!reader.isLastRead())
return false;
@@ -529,7 +550,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 13:
- ver = reader.readMessage("ver");
+ topVer = reader.readMessage("topVer");
if (!reader.isLastRead())
return false;
@@ -537,7 +558,15 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
reader.incrementState();
case 14:
- mvccSnapshot = reader.readMessage("mvccSnapshot");
+ txLbl = reader.readString("txLbl");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 15:
+ ver = reader.readMessage("ver");
if (!reader.isLastRead())
return false;
@@ -556,7 +585,7 @@ public class GridNearGetRequest extends GridCacheIdMessage implements GridCacheD
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 15;
+ return 16;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
index 2b3b5ea..ceddf72 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java
@@ -1121,7 +1121,8 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture<Bo
keepBinary,
clientFirst,
true,
- cctx.deploymentEnabled());
+ cctx.deploymentEnabled(),
+ inTx() ? tx.label() : null);
mapping.request(req);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
index f736cae..fb7295c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockRequest.java
@@ -86,6 +86,9 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
/** */
private byte flags;
+ /** Transaction label. */
+ private String txLbl;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
@@ -116,6 +119,7 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
* @param skipStore Skip store flag.
* @param firstClientReq {@code True} if first lock request for lock operation sent from client node.
* @param addDepInfo Deployment info flag.
+ * @param txLbl Transaction label.
*/
public GridNearLockRequest(
int cacheId,
@@ -141,7 +145,8 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
boolean keepBinary,
boolean firstClientReq,
boolean nearCache,
- boolean addDepInfo
+ boolean addDepInfo,
+ @Nullable String txLbl
) {
super(
cacheId,
@@ -169,6 +174,8 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
this.createTtl = createTtl;
this.accessTtl = accessTtl;
+ this.txLbl = txLbl;
+
dhtVers = new GridCacheVersion[keyCnt];
setFlag(syncCommit, SYNC_COMMIT_FLAG_MASK);
@@ -320,6 +327,13 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
return accessTtl;
}
+ /**
+ * @return Transaction label.
+ */
+ @Nullable public String txLabel() {
+ return txLbl;
+ }
+
/** {@inheritDoc} */
@Override public void prepareMarshal(GridCacheSharedContext ctx) throws IgniteCheckedException {
super.prepareMarshal(ctx);
@@ -417,6 +431,11 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
writer.incrementState();
+ case 29:
+ if(!writer.writeString("txLbl", txLbl))
+ return false;
+
+ writer.incrementState();
}
return true;
@@ -505,6 +524,14 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
reader.incrementState();
+ case 29:
+ txLbl = reader.readString("txLbl");
+
+ if(!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridNearLockRequest.class);
@@ -517,7 +544,7 @@ public class GridNearLockRequest extends GridDistributedLockRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 29;
+ return 30;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
index cf885e2..c87e521 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearSingleGetRequest.java
@@ -83,6 +83,9 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
/** TTL for read operation. */
private long accessTtl;
+ /** Transaction label. */
+ private @Nullable String txLbl;
+
/** */
private MvccSnapshot mvccSnapshot;
@@ -108,6 +111,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
* @param addReader Add reader flag.
* @param needVer {@code True} if entry version is needed.
* @param addDepInfo Deployment info.
+ * @param txLbl Transaction label.
* @param mvccSnapshot MVCC snapshot.
*/
public GridNearSingleGetRequest(
@@ -125,6 +129,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
boolean needVer,
boolean addDepInfo,
boolean recovery,
+ @Nullable String txLbl,
MvccSnapshot mvccSnapshot
) {
assert key != null;
@@ -138,6 +143,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
this.createTtl = createTtl;
this.accessTtl = accessTtl;
this.addDepInfo = addDepInfo;
+ this.txLbl = txLbl;
this.mvccSnapshot = mvccSnapshot;
if (readThrough)
@@ -222,6 +228,15 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
}
/**
+ * Get transaction label (may be null).
+ *
+ * @return Transaction label;
+ */
+ @Nullable public String txLabel() {
+ return txLbl;
+ }
+
+ /**
* @return Read through flag.
*/
public boolean readThrough() {
@@ -337,7 +352,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
reader.incrementState();
case 8:
- subjId = reader.readUuid("subjId");
+ mvccSnapshot = reader.readMessage("mvccSnapshot");
if (!reader.isLastRead())
return false;
@@ -345,7 +360,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
reader.incrementState();
case 9:
- taskNameHash = reader.readInt("taskNameHash");
+ subjId = reader.readUuid("subjId");
if (!reader.isLastRead())
return false;
@@ -353,7 +368,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
reader.incrementState();
case 10:
- topVer = reader.readMessage("topVer");
+ taskNameHash = reader.readInt("taskNameHash");
if (!reader.isLastRead())
return false;
@@ -361,7 +376,15 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
reader.incrementState();
case 11:
- mvccSnapshot = reader.readMessage("mvccSnapshot");
+ topVer = reader.readMessage("topVer");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 12:
+ txLbl = reader.readString("txLbl");
if (!reader.isLastRead())
return false;
@@ -419,25 +442,31 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
writer.incrementState();
case 8:
- if (!writer.writeUuid("subjId", subjId))
+ if (!writer.writeMessage("mvccSnapshot", mvccSnapshot))
return false;
writer.incrementState();
case 9:
- if (!writer.writeInt("taskNameHash", taskNameHash))
+ if (!writer.writeUuid("subjId", subjId))
return false;
writer.incrementState();
case 10:
- if (!writer.writeMessage("topVer", topVer))
+ if (!writer.writeInt("taskNameHash", taskNameHash))
return false;
writer.incrementState();
case 11:
- if (!writer.writeMessage("mvccSnapshot", mvccSnapshot))
+ if (!writer.writeMessage("topVer", topVer))
+ return false;
+
+ writer.incrementState();
+
+ case 12:
+ if (!writer.writeString("txLbl", txLbl))
return false;
writer.incrementState();
@@ -459,7 +488,7 @@ public class GridNearSingleGetRequest extends GridCacheIdMessage implements Grid
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 12;
+ return 13;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
index 3f7e48a..5cf4755 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTransactionalCache.java
@@ -329,7 +329,8 @@ public class GridNearTransactionalCache<K, V> extends GridNearCacheAdapter<K, V>
req.timeout(),
req.txSize(),
req.subjectId(),
- req.taskNameHash()
+ req.taskNameHash(),
+ req.txLabel()
);
tx = ctx.tm().onCreated(null, tx);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxEnlistFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxEnlistFuture.java
index ef5b898..2a7c2e4 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxEnlistFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxEnlistFuture.java
@@ -370,7 +370,8 @@ public class GridNearTxEnlistFuture extends GridNearTxAbstractEnlistFuture<GridC
-1,
this.tx.subjectId(),
this.tx.taskNameHash(),
- false);
+ false,
+ null);
dhtTx.mvccSnapshot(new MvccSnapshotWithoutTxs(mvccSnapshot.coordinatorVersion(),
mvccSnapshot.counter(), MVCC_OP_COUNTER_NA, mvccSnapshot.cleanupVersion()));
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
index d4e6ca4..f7c6000 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java
@@ -3067,7 +3067,8 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
needVer,
/*keepCacheObject*/true,
recovery,
- mvccReadSnapshot(cacheCtx)
+ mvccReadSnapshot(cacheCtx),
+ label()
).chain(new C1<IgniteInternalFuture<Object>, Void>() {
@Override public Void apply(IgniteInternalFuture<Object> f) {
try {
@@ -3099,6 +3100,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
skipVals,
needVer,
/*keepCacheObject*/true,
+ label(),
mvccReadSnapshot(cacheCtx)
).chain(new C1<IgniteInternalFuture<Map<Object, Object>>, Void>() {
@Override public Void apply(IgniteInternalFuture<Map<Object, Object>> f) {
@@ -4770,10 +4772,8 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter implements GridTimeou
return startTime() + timeout();
}
- /**
- * @return Tx label.
- */
- public String label() {
+ /** {@inheritDoc} */
+ @Override public String label() {
return lb;
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
index 55c809d..d740737 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareRequest.java
@@ -28,6 +28,7 @@ import org.apache.ignite.internal.processors.cache.GridCacheContext;
import org.apache.ignite.internal.processors.cache.distributed.GridDistributedTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.util.tostring.GridToStringExclude;
+import org.apache.ignite.internal.util.tostring.GridToStringInclude;
import org.apache.ignite.internal.util.typedef.F;
import org.apache.ignite.internal.util.typedef.internal.S;
import org.apache.ignite.internal.util.typedef.internal.U;
@@ -80,6 +81,10 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
@GridToStringExclude
private byte flags;
+ /** Transaction label. */
+ @GridToStringInclude
+ @Nullable private String txLbl;
+
/**
* Empty constructor required for {@link Externalizable}.
*/
@@ -145,6 +150,8 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
this.subjId = subjId;
this.taskNameHash = taskNameHash;
+ txLbl = tx.label();
+
setFlag(near, NEAR_FLAG_MASK);
setFlag(implicitSingle, IMPLICIT_SINGLE_FLAG_MASK);
setFlag(explicitLock, EXPLICIT_LOCK_FLAG_MASK);
@@ -245,6 +252,13 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
}
/**
+ * @return Transaction label.
+ */
+ @Nullable public String txLabel() {
+ return txLbl;
+ }
+
+ /**
*
*/
public void cloneEntries() {
@@ -354,6 +368,12 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
writer.incrementState();
+ case 26:
+ if (!writer.writeString("txLbl", txLbl))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -418,6 +438,14 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
+ case 26:
+ txLbl = reader.readString("txLbl");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridNearTxPrepareRequest.class);
@@ -430,7 +458,7 @@ public class GridNearTxPrepareRequest extends GridDistributedTxPrepareRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 26;
+ return 27;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java
index 74b6fda..d6deed1 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxQueryResultsEnlistFuture.java
@@ -350,7 +350,8 @@ public class GridNearTxQueryResultsEnlistFuture extends GridNearTxQueryAbstractE
-1,
this.tx.subjectId(),
this.tx.taskNameHash(),
- false);
+ false,
+ tx.label());
dhtTx.mvccSnapshot(new MvccSnapshotWithoutTxs(mvccSnapshot.coordinatorVersion(),
mvccSnapshot.counter(), MVCC_OP_COUNTER_NA, mvccSnapshot.cleanupVersion()));
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
index 879bf26..c72dd73 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxRemote.java
@@ -90,6 +90,7 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
* @param txSize Expected transaction size.
* @param subjId Subject ID.
* @param taskNameHash Task name hash code.
+ * @param txLbl Transaction label.
* @throws IgniteCheckedException If unmarshalling failed.
*/
public GridNearTxRemote(
@@ -109,22 +110,24 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
Collection<IgniteTxEntry> writeEntries,
int txSize,
@Nullable UUID subjId,
- int taskNameHash
+ int taskNameHash,
+ @Nullable String txLbl
) throws IgniteCheckedException {
super(
- ctx,
- nodeId,
+ ctx,
+ nodeId,
xidVer,
commitVer,
- sys,
- plc,
- concurrency,
- isolation,
- invalidate,
+ sys,
+ plc,
+ concurrency,
+ isolation,
+ invalidate,
timeout,
txSize,
- subjId,
- taskNameHash
+ subjId,
+ taskNameHash,
+ txLbl
);
assert nearNodeId != null;
@@ -168,6 +171,7 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
* @param txSize Expected transaction size.
* @param subjId Subject ID.
* @param taskNameHash Task name hash code.
+ * @param txLbl Transaction label.
*/
public GridNearTxRemote(
GridCacheSharedContext ctx,
@@ -185,22 +189,24 @@ public class GridNearTxRemote extends GridDistributedTxRemoteAdapter {
long timeout,
int txSize,
@Nullable UUID subjId,
- int taskNameHash
+ int taskNameHash,
+ @Nullable String txLbl
) {
super(
- ctx,
- nodeId,
+ ctx,
+ nodeId,
xidVer,
commitVer,
sys,
plc,
- concurrency,
- isolation,
- invalidate,
+ concurrency,
+ isolation,
+ invalidate,
timeout,
txSize,
subjId,
- taskNameHash
+ taskNameHash,
+ txLbl
);
assert nearNodeId != null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
index 4cacb47..919f6d2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/local/atomic/GridLocalAtomicCache.java
@@ -427,6 +427,7 @@ public class GridLocalAtomicCache<K, V> extends GridLocalCache<K, V> {
if (evt) {
ctx.events().readEvent(cacheKey,
null,
+ null,
row.value(),
subjId,
taskName,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
index 05ebe5d..7588fb0 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java
@@ -642,6 +642,13 @@ public interface IgniteInternalTx {
public void commitError(Throwable e);
/**
+ * Returns label of transactions.
+ *
+ * @return Label of transaction or {@code null} if there was not set.
+ */
+ @Nullable public String label();
+
+ /**
* @param mvccSnapshot Mvcc snapshot.
*/
public void mvccSnapshot(MvccSnapshot mvccSnapshot);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java
----------------------------------------------------------------------
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 f973b89..8b24e01 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
@@ -2232,6 +2232,11 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter implement
}
/** {@inheritDoc} */
+ @Nullable @Override public String label() {
+ throw new IllegalStateException("Deserialized transaction can only be used as read-only.");
+ }
+
+ /** {@inheritDoc} */
@Override public boolean empty() {
throw new IllegalStateException("Deserialized transaction can only be used as read-only.");
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxHandler.java
----------------------------------------------------------------------
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 b872047..93a101b 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
@@ -509,7 +509,8 @@ public class IgniteTxHandler {
req.txSize(),
req.transactionNodes(),
req.subjectId(),
- req.taskNameHash()
+ req.taskNameHash(),
+ req.txLabel()
);
tx = ctx.tm().onCreated(null, tx);
@@ -1682,7 +1683,8 @@ public class IgniteTxHandler {
req.subjectId(),
req.taskNameHash(),
single,
- req.storeWriteThrough());
+ req.storeWriteThrough(),
+ req.txLabel());
tx.onePhaseCommit(req.onePhaseCommit());
tx.writeVersion(req.writeVersion());
@@ -2023,7 +2025,8 @@ public class IgniteTxHandler {
req.nearWrites(),
req.txSize(),
req.subjectId(),
- req.taskNameHash()
+ req.taskNameHash(),
+ req.txLabel()
);
tx.writeVersion(req.writeVersion());
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CashEventWithTxLabelTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CashEventWithTxLabelTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CashEventWithTxLabelTest.java
new file mode 100644
index 0000000..da25288
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CashEventWithTxLabelTest.java
@@ -0,0 +1,487 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.processors.cache;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntryProcessor;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.events.CacheEvent;
+import org.apache.ignite.events.Event;
+import org.apache.ignite.internal.IgniteKernal;
+import org.apache.ignite.internal.util.lang.IgnitePair;
+import org.apache.ignite.lang.IgnitePredicate;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionConcurrency;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Assert;
+
+import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_PUT;
+import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_READ;
+import static org.apache.ignite.events.EventType.EVT_CACHE_OBJECT_REMOVED;
+
+/**
+ * Test to check passing transaction's label for EVT_CACHE_OBJECT_READ, EVT_CACHE_OBJECT_PUT,
+ * EVT_CACHE_OBJECT_REMOVED events.
+ */
+public class CashEventWithTxLabelTest extends GridCommonAbstractTest {
+ /** Types event to be checked. */
+ private static final int[] CACHE_EVENT_TYPES = {EVT_CACHE_OBJECT_READ, EVT_CACHE_OBJECT_PUT, EVT_CACHE_OBJECT_REMOVED};
+
+ /** Transaction label. */
+ private static final String TX_LABEL = "TX_LABEL";
+
+ /** Number of server nodes. */
+ private static final int SRVS = 3;
+
+ /** Number of client nodes. */
+ private static final int CLIENTS = 1;
+
+ /** Cache name. */
+ public static final String CACHE_NAME = "cache";
+
+ /** Client or server mode to start Ignite instance. */
+ private static boolean client;
+
+ /** Key related to primary node. */
+ private Integer primaryKey = 0;
+
+ /** Key related to backup node. */
+ private Integer backupKey = 0;
+
+ /** Current cash backup count. */
+ private int backupCnt;
+
+ /** Current transaction isolation level. */
+ private TransactionIsolation isolation;
+
+ /** Current transaction concurrency level. */
+ private TransactionConcurrency concurrency;
+
+ /** All failed tests information, */
+ private ArrayList<String> errors = new ArrayList<>();
+
+ /** Count of errors on previous iteration of testing. */
+ private int prevErrCnt = 0;
+
+ /** List to keep all events with no tx label between run tests */
+ private static List<CacheEvent> wrongEvts = Collections.synchronizedList(new ArrayList<>());
+
+ /** Simple entry processor to use for tests */
+ private static CacheEntryProcessor entryProcessor = (CacheEntryProcessor)(entry, objects) -> entry.getValue();
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName).setClientMode(client);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ client = false;
+
+ startGridsMultiThreaded(SRVS);
+
+ client = true;
+
+ startGridsMultiThreaded(SRVS, CLIENTS);
+
+ client = false;
+
+ waitForDiscovery(primary(), backup1(), backup2(), client());
+
+ registerEventListeners(primary(), backup1(), backup2(), client());
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ super.afterTestsStopped();
+
+ stopAllGrids();
+ }
+
+ /**
+ * Check all cases for passing transaction label in cash event.
+ *
+ * @throws Exception If failed.
+ */
+ public void testPassTxLabelInCashEventForAllCases() throws Exception {
+ Ignite[] nodes = {client(), primary(), backup1(), backup2()};
+
+ for (int backupCnt = 0; backupCnt < SRVS; backupCnt++) {
+ this.backupCnt = backupCnt;
+
+ prepareCache(backupCnt);
+
+ for (TransactionIsolation isolation : TransactionIsolation.values()) {
+ this.isolation = isolation;
+
+ for (TransactionConcurrency concurrency : TransactionConcurrency.values()) {
+ this.concurrency = concurrency;
+
+ for (int i = 0; i < nodes.length - 1; i++) {
+ Ignite nodeForPut = nodes[i];
+ Ignite nodeForGet = nodes[i + 1];
+
+ singleWriteReadRemoveTest(nodeForPut, nodeForGet);
+
+ multiWriteReadRemoveTest(nodeForPut, nodeForGet);
+
+ singleNodeBatchWriteReadRemoveTest(nodeForPut, nodeForGet);
+
+ multiNodeBatchWriteReadRemoveTest(nodeForPut, nodeForGet);
+
+ writeInvokeRemoveTest(nodeForPut, nodeForGet);
+
+ writeInvokeAllRemoveTest(nodeForPut, nodeForGet);
+ }
+ }
+ }
+ }
+
+ String listOfFailedTests = String.join(",\n", errors);
+
+ Assert.assertTrue("Have been received " + prevErrCnt + " cache events with incorrect txlabel.\n" +
+ "Failed tests:" + listOfFailedTests,
+ errors.isEmpty());
+ }
+
+ /**
+ * Check error after run test. In case error occured information about failed test will be added to errors list.
+ *
+ * @param testName Name of test which result will be checked.
+ * @param node1 First node
+ * @param node2 Second node
+ */
+ private void checkResult(String testName, Ignite node1, Ignite node2) {
+ int currErrCnt = wrongEvts.size();
+
+ if (prevErrCnt != currErrCnt) {
+ prevErrCnt = currErrCnt;
+
+ errors.add(String.format("%s backCnt-%s, %s, %s, node1-%s, node2-%s",
+ testName, backupCnt, isolation, concurrency, nodeType(node1), nodeType(node2)));
+ }
+ }
+
+ /**
+ * @param node Ignite node
+ * @return Node type in the test
+ */
+ private String nodeType(Ignite node) {
+ if (client().equals(node))
+ return "CLIENT";
+ else if (primary().equals(node))
+ return "PRIMARY";
+ else if (backup1().equals(node))
+ return "BACKUP1";
+ else if (backup2().equals(node))
+ return "BACKUP2";
+ else
+ return "UNKNOWN";
+ }
+
+ /**
+ * Test single put, get, remove operations.
+ *
+ * @param instanceToPut Ignite instance to put test data.
+ * @param instanceToGet Ignite instance to get test data.
+ */
+ private void singleWriteReadRemoveTest(Ignite instanceToPut, Ignite instanceToGet) {
+ runTransactionally(instanceToPut, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).put(primaryKey, 3);
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).get(primaryKey);
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).remove(primaryKey);
+ });
+
+ checkResult("singleWriteReadRemoveTest", instanceToPut, instanceToGet);
+ }
+
+ /**
+ * Test multi put, get, remove operations
+ *
+ * @param instanceToPut Ignite instance to put test data.
+ * @param instanceToGet Ignite instance to get test data.
+ */
+ private void multiWriteReadRemoveTest(Ignite instanceToPut, Ignite instanceToGet) {
+ runTransactionally(instanceToPut, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).put(primaryKey, 2);
+ ign.cache(CACHE_NAME).put(backupKey, 3);
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).get(primaryKey);
+ ign.cache(CACHE_NAME).get(backupKey);
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).remove(primaryKey);
+ ign.cache(CACHE_NAME).remove(backupKey);
+ });
+
+ checkResult("multiWriteReadRemoveTest", instanceToPut, instanceToGet);
+ }
+
+ /**
+ * Test multi nodes batch write-read
+ *
+ * @param instanceToPut Ignite instance to put test data.
+ * @param instanceToGet Ignite instance to get test data.
+ */
+ private void multiNodeBatchWriteReadRemoveTest(Ignite instanceToPut, Ignite instanceToGet) {
+ Map<Integer, Integer> keyValuesMap = IntStream.range(0, 100).boxed()
+ .collect(Collectors.toMap(Function.identity(), Function.identity()));
+
+ runTransactionally(instanceToPut, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).putAll(keyValuesMap);
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).getAll(keyValuesMap.keySet());
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).removeAll(keyValuesMap.keySet());
+ });
+
+ checkResult("multiNodeBatchWriteReadRemoveTest", instanceToPut, instanceToGet);
+ }
+
+ /**
+ * Test single node batch write-read-remove
+ *
+ * @param instanceToPut Ignite instance to put test data.
+ * @param instanceToGet Ignite instance to get test data.
+ */
+ private void singleNodeBatchWriteReadRemoveTest(Ignite instanceToPut, Ignite instanceToGet) {
+ IgnitePair<Integer> keys = evaluatePrimaryAndBackupKeys(primaryKey + 1, backupKey + 1);
+
+ Map<Integer, Integer> keyValuesMap = new HashMap<>();
+ keyValuesMap.put(primaryKey, 1);
+ keyValuesMap.put(keys.get1(), 2);
+
+ runTransactionally(instanceToPut, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).putAll(keyValuesMap);
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).getAll(keyValuesMap.keySet());
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).removeAll(keyValuesMap.keySet());
+ });
+
+ checkResult("oneNodeBatchWriteReadRemoveTest", instanceToPut, instanceToGet);
+ }
+
+ /**
+ * Test put-invoke-remove
+ *
+ * @param instanceToPut Ignite instance to put test data.
+ * @param instanceToGet Ignite instance to get test data.
+ */
+ @SuppressWarnings("unchecked")
+ private void writeInvokeRemoveTest(Ignite instanceToPut, Ignite instanceToGet) {
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).put(primaryKey, 3);
+ });
+
+ runTransactionally(instanceToPut, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).invoke(primaryKey, entryProcessor);
+ ign.cache(CACHE_NAME).invoke(backupKey, entryProcessor);
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).remove(primaryKey);
+ });
+
+ checkResult("writeInvokeRemoveTest", instanceToPut, instanceToGet);
+ }
+
+ /**
+ * Test putAll-invokeAll-removeAll
+ *
+ * @param instanceToPut Ignite instance to put test data.
+ * @param instanceToGet Ignite instance to get test data.
+ */
+ @SuppressWarnings("unchecked")
+ private void writeInvokeAllRemoveTest(Ignite instanceToPut, Ignite instanceToGet) {
+ Map<Integer, Integer> keyValuesMap = IntStream.range(0, 100).boxed()
+ .collect(Collectors.toMap(Function.identity(), Function.identity()));
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).putAll(keyValuesMap);
+ });
+
+ runTransactionally(instanceToPut, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).invokeAll(keyValuesMap.keySet(), entryProcessor);
+ });
+
+ runTransactionally(instanceToGet, (Ignite ign) -> {
+ ign.cache(CACHE_NAME).removeAll(keyValuesMap.keySet());
+ });
+
+ checkResult("WriteInvokeAllRemoveTest", instanceToPut, instanceToGet);
+ }
+
+ /**
+ * Run command in transaction.
+ *
+ * @param startNode Ignite node to start transaction and run passed command.
+ * @param cmdInTx Command which should be done in transaction.
+ */
+ private void runTransactionally(Ignite startNode, Consumer<Ignite> cmdInTx) {
+ try (Transaction tx = startNode.transactions().withLabel(TX_LABEL).txStart(concurrency, isolation)) {
+ cmdInTx.accept(startNode);
+
+ tx.commit();
+ }
+ }
+
+ /**
+ * Add event listener to passed Ignite instances for cache event types.
+ *
+ * @param igns Ignite instances.
+ */
+ private void registerEventListeners(Ignite... igns) {
+ if (igns != null) {
+ for (Ignite ign : igns) {
+ ign.events().enableLocal(CACHE_EVENT_TYPES);
+ ign.events().localListen((IgnitePredicate<Event>)event -> {
+ CacheEvent cacheEvt = (CacheEvent)event;
+
+ if (!TX_LABEL.equals(cacheEvt.txLabel())) {
+ log.error("Has been received event with incorrect label " + cacheEvt.txLabel() + " ," +
+ " expected " + TX_LABEL + " label");
+
+ wrongEvts.add(cacheEvt);
+ }
+
+ return true;
+ }, CACHE_EVENT_TYPES);
+ }
+ }
+ }
+
+ /**
+ * Create cache with passed number of backups and determinate primary and backup keys. If cache was created before
+ * it will be removed before create new one.
+ *
+ * @param cacheBackups Number of backups for cache.
+ * @throws InterruptedException In case of fail.
+ */
+ private void prepareCache(int cacheBackups) throws InterruptedException {
+ IgniteCache<Object, Object> cache = client().cache(CACHE_NAME);
+
+ if (cache != null)
+ cache.destroy();
+
+ client().createCache(
+ new CacheConfiguration<Integer, Integer>()
+ .setName(CACHE_NAME)
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ .setCacheMode(CacheMode.PARTITIONED)
+ .setBackups(cacheBackups)
+ );
+
+ awaitPartitionMapExchange();
+
+ IgnitePair<Integer> keys = evaluatePrimaryAndBackupKeys(0, 0);
+
+ primaryKey = keys.get1();
+ backupKey = keys.get2();
+ }
+
+ /**
+ * Evaluate primary and backup keys.
+ *
+ * @param primaryKeyStart Value from need to start calculate primary key.
+ * @param backupKeyStart Value from need to start calculate backup key.
+ * @return Pair of result. The first result is found primary key. The second is found backup key.
+ */
+ private IgnitePair<Integer> evaluatePrimaryAndBackupKeys(final int primaryKeyStart, final int backupKeyStart) {
+ int primaryKey = primaryKeyStart;
+ int backupKey = backupKeyStart;
+
+ while (!client().affinity(CACHE_NAME).isPrimary(((IgniteKernal)primary()).localNode(), primaryKey))
+ primaryKey++;
+
+ while (!client().affinity(CACHE_NAME).isBackup(((IgniteKernal)primary()).localNode(), backupKey)
+ && backupKey < 100 + backupKeyStart)
+ backupKey++;
+
+ return new IgnitePair<>(primaryKey, backupKey);
+ }
+
+ /**
+ * Return primary node.
+ *
+ * @return Primary node.
+ */
+ private Ignite primary() {
+ return ignite(0);
+ }
+
+ /**
+ * Return first backup node.
+ *
+ * @return First backup node.
+ */
+ private Ignite backup1() {
+ return ignite(1);
+ }
+
+ /**
+ * Return second backup node.
+ *
+ * @return Second backup node.
+ */
+ private Ignite backup2() {
+ return ignite(2);
+ }
+
+ /**
+ * Return client node.
+ *
+ * @return Client node.
+ */
+ private Ignite client() {
+ return ignite(3);
+ }
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccOperationChecksTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccOperationChecksTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccOperationChecksTest.java
index e9083f7..49196d2 100644
--- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccOperationChecksTest.java
+++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/mvcc/CacheMvccOperationChecksTest.java
@@ -26,7 +26,6 @@ import javax.cache.expiry.ExpiryPolicy;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
import org.apache.ignite.cache.CacheMode;
-import org.apache.ignite.cache.CachePeekMode;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.processors.cache.GridCacheAdapter;
import org.apache.ignite.internal.util.typedef.internal.U;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/test/java/org/apache/ignite/platform/PlatformEventsWriteEventTask.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/platform/PlatformEventsWriteEventTask.java b/modules/core/src/test/java/org/apache/ignite/platform/PlatformEventsWriteEventTask.java
index 2d3599b..0c626bb 100644
--- a/modules/core/src/test/java/org/apache/ignite/platform/PlatformEventsWriteEventTask.java
+++ b/modules/core/src/test/java/org/apache/ignite/platform/PlatformEventsWriteEventTask.java
@@ -92,7 +92,7 @@ public class PlatformEventsWriteEventTask extends ComputeTaskAdapter<Long, Objec
IgniteUuid igniteUuid = new IgniteUuid(uuid, 3);
ctx.writeEvent(writer, new CacheEvent("cacheName", node, node, "msg", evtType, 1, true, 2,
- igniteUuid, 3, 4, true, 5, true, uuid, "cloClsName", "taskName"));
+ igniteUuid, "txLabel",3, 4, true, 5, true, uuid, "cloClsName", "taskName"));
//noinspection unchecked
ctx.writeEvent(writer, new CacheQueryExecutedEvent(node, msg, evtType, "qryType", "cacheName",
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
index 8ea329e..5f7377f 100644
--- a/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
+++ b/modules/core/src/test/java/org/apache/ignite/testsuites/IgniteCacheTestSuite4.java
@@ -46,6 +46,7 @@ import org.apache.ignite.internal.processors.cache.CacheStoreUsageMultinodeDynam
import org.apache.ignite.internal.processors.cache.CacheStoreUsageMultinodeStaticStartAtomicTest;
import org.apache.ignite.internal.processors.cache.CacheStoreUsageMultinodeStaticStartTxTest;
import org.apache.ignite.internal.processors.cache.CacheTxNotAllowReadFromBackupTest;
+import org.apache.ignite.internal.processors.cache.CashEventWithTxLabelTest;
import org.apache.ignite.internal.processors.cache.CrossCacheLockTest;
import org.apache.ignite.internal.processors.cache.GridCacheMarshallingNodeJoinSelfTest;
import org.apache.ignite.internal.processors.cache.GridCacheMultinodeUpdateAtomicNearEnabledSelfTest;
@@ -340,6 +341,8 @@ public class IgniteCacheTestSuite4 extends TestSuite {
suite.addTestSuite(CacheResultIsNotNullOnPartitionLossTest.class);
+ suite.addTestSuite(CashEventWithTxLabelTest.class);
+
return suite;
}
}
[2/2] ignite git commit: IGNITE-9274: Added optional labels to
transactions. This closes #4600.
Posted by vo...@apache.org.
IGNITE-9274: Added optional labels to transactions. This closes #4600.
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/3c54f389
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/3c54f389
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/3c54f389
Branch: refs/heads/master
Commit: 3c54f389415ae317c6d5e8140fc4d23c39ac51db
Parents: d37f1d7
Author: Yuriy Gerzhedovich <yg...@gridgain.com>
Authored: Thu Nov 15 12:08:54 2018 +0300
Committer: devozerov <vo...@gridgain.com>
Committed: Thu Nov 15 12:08:54 2018 +0300
----------------------------------------------------------------------
.../ignite/codegen/MessageCodeGenerator.java | 11 +-
.../org/apache/ignite/events/CacheEvent.java | 24 +-
.../processors/cache/CacheGroupContext.java | 3 +-
.../cache/ExchangeDiscoveryEvents.java | 1 +
.../processors/cache/GridCacheAdapter.java | 4 +
.../cache/GridCacheConcurrentMapImpl.java | 9 +-
.../processors/cache/GridCacheEventManager.java | 29 +-
.../cache/GridCacheEvictionManager.java | 4 +-
.../processors/cache/GridCacheMapEntry.java | 18 +-
.../GridDistributedTxRemoteAdapter.java | 14 +-
.../distributed/dht/GridDhtCacheAdapter.java | 11 +
.../cache/distributed/dht/GridDhtGetFuture.java | 8 +
.../distributed/dht/GridDhtGetSingleFuture.java | 8 +
.../distributed/dht/GridDhtLockFuture.java | 7 +-
.../distributed/dht/GridDhtLockRequest.java | 32 +-
.../dht/GridDhtTransactionalCacheAdapter.java | 12 +-
.../cache/distributed/dht/GridDhtTxLocal.java | 14 +-
.../distributed/dht/GridDhtTxPrepareFuture.java | 4 +-
.../dht/GridDhtTxPrepareRequest.java | 89 ++--
.../cache/distributed/dht/GridDhtTxRemote.java | 14 +-
.../dht/GridPartitionedGetFuture.java | 10 +
.../dht/GridPartitionedSingleGetFuture.java | 10 +
.../dht/atomic/GridDhtAtomicCache.java | 3 +
.../dht/colocated/GridDhtColocatedCache.java | 55 ++-
.../colocated/GridDhtColocatedLockFuture.java | 3 +-
.../dht/preloader/GridDhtForceKeysFuture.java | 4 +-
.../dht/preloader/GridDhtPartitionDemander.java | 4 +-
.../dht/topology/GridDhtLocalPartition.java | 3 +-
.../distributed/near/GridNearGetFuture.java | 2 +
.../distributed/near/GridNearGetRequest.java | 55 ++-
.../distributed/near/GridNearLockFuture.java | 3 +-
.../distributed/near/GridNearLockRequest.java | 31 +-
.../near/GridNearSingleGetRequest.java | 47 +-
.../near/GridNearTransactionalCache.java | 3 +-
.../near/GridNearTxEnlistFuture.java | 3 +-
.../cache/distributed/near/GridNearTxLocal.java | 10 +-
.../near/GridNearTxPrepareRequest.java | 30 +-
.../GridNearTxQueryResultsEnlistFuture.java | 3 +-
.../distributed/near/GridNearTxRemote.java | 40 +-
.../local/atomic/GridLocalAtomicCache.java | 1 +
.../cache/transactions/IgniteInternalTx.java | 7 +
.../cache/transactions/IgniteTxAdapter.java | 5 +
.../cache/transactions/IgniteTxHandler.java | 9 +-
.../cache/CashEventWithTxLabelTest.java | 487 +++++++++++++++++++
.../mvcc/CacheMvccOperationChecksTest.java | 1 -
.../platform/PlatformEventsWriteEventTask.java | 2 +-
.../testsuites/IgniteCacheTestSuite4.java | 3 +
47 files changed, 1014 insertions(+), 136 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
----------------------------------------------------------------------
diff --git a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
index 7492e51..615ffca 100644
--- a/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
+++ b/modules/codegen/src/main/java/org/apache/ignite/codegen/MessageCodeGenerator.java
@@ -43,6 +43,11 @@ import org.apache.ignite.internal.GridDirectCollection;
import org.apache.ignite.internal.GridDirectMap;
import org.apache.ignite.internal.GridDirectTransient;
import org.apache.ignite.internal.IgniteCodeGeneratingFail;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtLockRequest;
+import org.apache.ignite.internal.processors.cache.distributed.dht.GridDhtTxPrepareRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearSingleGetRequest;
+import org.apache.ignite.internal.processors.cache.distributed.near.GridNearTxPrepareRequest;
import org.apache.ignite.internal.processors.cache.mvcc.msg.MvccRecoveryFinishedMessage;
import org.apache.ignite.internal.util.IgniteUtils;
import org.apache.ignite.internal.util.typedef.internal.SB;
@@ -169,7 +174,11 @@ public class MessageCodeGenerator {
// gen.generateAll(true);
- gen.generateAndWrite(MvccRecoveryFinishedMessage.class);
+ gen.generateAndWrite(GridNearGetRequest.class);
+ gen.generateAndWrite(GridNearSingleGetRequest.class);
+ gen.generateAndWrite(GridNearTxPrepareRequest.class);
+ gen.generateAndWrite(GridDhtTxPrepareRequest.class);
+ gen.generateAndWrite(GridDhtLockRequest.class);
// gen.generateAndWrite(GridNearAtomicUpdateRequest.class);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/events/CacheEvent.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/events/CacheEvent.java b/modules/core/src/main/java/org/apache/ignite/events/CacheEvent.java
index 5aa9d06..92b2ade 100644
--- a/modules/core/src/main/java/org/apache/ignite/events/CacheEvent.java
+++ b/modules/core/src/main/java/org/apache/ignite/events/CacheEvent.java
@@ -90,10 +90,14 @@ public class CacheEvent extends EventAdapter {
@GridToStringInclude(sensitive = true)
private Object key;
- /** Event ID. */
+ /** Transaction Id. */
@GridToStringInclude
private final IgniteUuid xid;
+ /** Transaction label. */
+ @GridToStringInclude
+ private String txLbl;
+
/** Lock ID. */
@GridToStringInclude
private final Object lockId;
@@ -152,6 +156,7 @@ public class CacheEvent extends EventAdapter {
* @param near Flag indicating whether event happened on {@code near} or {@code partitioned} cache.
* @param key Cache key.
* @param xid Transaction ID.
+ * @param txLbl Transaction label.
* @param lockId Lock ID.
* @param newVal New value.
* @param hasNewVal Flag indicating whether new value is present in case if we
@@ -163,7 +168,7 @@ public class CacheEvent extends EventAdapter {
* @param cloClsName Closure class name.
*/
public CacheEvent(String cacheName, ClusterNode node, @Nullable ClusterNode evtNode, String msg, int type, int part,
- boolean near, Object key, IgniteUuid xid, Object lockId, Object newVal, boolean hasNewVal,
+ boolean near, Object key, IgniteUuid xid, String txLbl, Object lockId, Object newVal, boolean hasNewVal,
Object oldVal, boolean hasOldVal, UUID subjId, String cloClsName, String taskName) {
super(node, msg, type);
this.cacheName = cacheName;
@@ -172,6 +177,7 @@ public class CacheEvent extends EventAdapter {
this.near = near;
this.key = key;
this.xid = xid;
+ this.txLbl = txLbl;
this.lockId = lockId;
this.newVal = newVal;
this.hasNewVal = hasNewVal;
@@ -229,7 +235,7 @@ public class CacheEvent extends EventAdapter {
}
/**
- * ID of surrounding cache cache transaction or <tt>null</tt> if there is
+ * ID of surrounding cache transaction or <tt>null</tt> if there is
* no surrounding transaction.
*
* @return ID of surrounding cache transaction.
@@ -239,6 +245,16 @@ public class CacheEvent extends EventAdapter {
}
/**
+ * Label of surrounding cache transaction or <tt>null</tt> if there either is
+ * no surrounding transaction or label was not set.
+ *
+ * @return Label of surrounding cache transaction.
+ */
+ public String txLabel() {
+ return txLbl;
+ }
+
+ /**
* ID of the lock if held or <tt>null</tt> if no lock held.
*
* @return ID of the lock if held.
@@ -336,4 +352,4 @@ public class CacheEvent extends EventAdapter {
"type", name(),
"tstamp", timestamp());
}
-}
\ No newline at end of file
+}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
index 93ba69c..7f0b4ed 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/CacheGroupContext.java
@@ -512,7 +512,8 @@ public class CacheGroupContext {
cctx.events().addEvent(part,
key,
evtNodeId,
- (IgniteUuid)null,
+ null,
+ null,
null,
type,
newVal,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java
index 9434165..1d5d017 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/ExchangeDiscoveryEvents.java
@@ -222,6 +222,7 @@ public class ExchangeDiscoveryEvents {
null,
null,
null,
+ null,
false,
null,
false,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
index 18a4da4..feb609a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheAdapter.java
@@ -1926,6 +1926,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
/*keep cache objects*/false,
recovery,
needVer,
+ null,
null); // TODO IGNITE-7371
}
@@ -1941,6 +1942,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
* @param skipVals Skip values flag.
* @param keepCacheObjects Keep cache objects.
* @param needVer If {@code true} returns values as tuples containing value and version.
+ * @param txLbl Transaction label.
* @param mvccSnapshot MVCC snapshot.
* @return Future.
*/
@@ -1957,6 +1959,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
final boolean keepCacheObjects,
final boolean recovery,
final boolean needVer,
+ @Nullable String txLbl,
MvccSnapshot mvccSnapshot
) {
if (F.isEmpty(keys))
@@ -2042,6 +2045,7 @@ public abstract class GridCacheAdapter<K, V> implements IgniteInternalCache<K, V
if (evt) {
ctx.events().readEvent(key,
null,
+ txLbl,
row.value(),
subjId,
taskName,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMapImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMapImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMapImpl.java
index 75c0d0c..005fdfb 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMapImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheConcurrentMapImpl.java
@@ -168,7 +168,8 @@ public abstract class GridCacheConcurrentMapImpl implements GridCacheConcurrentM
ctx.events().addEvent(doomed.partition(),
doomed.key(),
ctx.localNodeId(),
- (IgniteUuid)null,
+ null,
+ null,
null,
EVT_CACHE_ENTRY_DESTROYED,
null,
@@ -188,7 +189,8 @@ public abstract class GridCacheConcurrentMapImpl implements GridCacheConcurrentM
ctx.events().addEvent(created.partition(),
created.key(),
ctx.localNodeId(),
- (IgniteUuid)null,
+ null,
+ null,
null,
EVT_CACHE_ENTRY_CREATED,
null,
@@ -274,7 +276,8 @@ public abstract class GridCacheConcurrentMapImpl implements GridCacheConcurrentM
ctx.events().addEvent(entry.partition(),
entry.key(),
ctx.localNodeId(),
- (IgniteUuid)null,
+ null,
+ null,
null,
EVT_CACHE_ENTRY_DESTROYED,
null,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEventManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEventManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEventManager.java
index ef35016..726a6c8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEventManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEventManager.java
@@ -63,6 +63,7 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
/**
* @param key Key for event.
* @param tx Possible surrounding transaction.
+ * @param txLbl Possible lable of possible surrounding transaction.
* @param val Read value.
* @param subjId Subject ID.
* @param taskName Task name.
@@ -70,6 +71,7 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
*/
public void readEvent(KeyCacheObject key,
@Nullable IgniteInternalTx tx,
+ @Nullable String txLbl,
@Nullable CacheObject val,
@Nullable UUID subjId,
@Nullable String taskName,
@@ -77,7 +79,9 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
if (isRecordable(EVT_CACHE_OBJECT_READ)) {
addEvent(cctx.affinity().partition(key),
key,
+ cctx.localNodeId(),
tx,
+ txLbl,
null,
EVT_CACHE_OBJECT_READ,
val,
@@ -107,7 +111,7 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
*/
public void addEvent(int part,
KeyCacheObject key,
- IgniteInternalTx tx,
+ @Nullable IgniteInternalTx tx,
@Nullable GridCacheMvccCandidate owner,
int type,
@Nullable CacheObject newVal,
@@ -143,7 +147,8 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
0,
null,
cctx.localNodeId(),
- (IgniteUuid)null,
+ null,
+ null,
null,
type,
null,
@@ -174,7 +179,7 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
public void addEvent(int part,
KeyCacheObject key,
UUID nodeId,
- IgniteInternalTx tx,
+ @Nullable IgniteInternalTx tx,
GridCacheMvccCandidate owner,
int type,
CacheObject newVal,
@@ -188,7 +193,9 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
{
addEvent(part,
key,
- nodeId, tx == null ? null : tx.xid(),
+ nodeId,
+ tx,
+ null,
owner == null ? null : owner.version(),
type,
newVal,
@@ -234,7 +241,8 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
addEvent(part,
key,
evtNodeId,
- tx == null ? null : tx.xid(),
+ tx,
+ null,
owner == null ? null : owner.version(),
type,
newVal,
@@ -251,7 +259,8 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
* @param part Partition.
* @param key Key for the event.
* @param evtNodeId Event node ID.
- * @param xid Transaction ID.
+ * @param tx Possible surrounding transaction.
+ * @param txLbl Possible label of possible surrounding transaction.
* @param lockId Lock ID.
* @param type Event type.
* @param newVal New value.
@@ -266,7 +275,8 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
int part,
KeyCacheObject key,
UUID evtNodeId,
- @Nullable IgniteUuid xid,
+ @Nullable IgniteInternalTx tx,
+ @Nullable String txLbl,
@Nullable Object lockId,
int type,
@Nullable CacheObject newVal,
@@ -324,6 +334,10 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
oldVal0 = cctx.cacheObjectContext().unwrapBinaryIfNeeded(oldVal, true, false);
}
+ IgniteUuid xid = tx == null ? null : tx.xid();
+
+ String finalTxLbl = (tx == null || tx.label() == null) ? txLbl : tx.label();
+
cctx.gridEvents().record(new CacheEvent(cctx.name(),
cctx.localNode(),
evtNode,
@@ -333,6 +347,7 @@ public class GridCacheEventManager extends GridCacheManagerAdapter {
cctx.isNear(),
key0,
xid,
+ finalTxLbl,
lockId,
val0,
hasNewVal,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
index cba537a..b3b8810 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheEvictionManager.java
@@ -155,7 +155,7 @@ public class GridCacheEvictionManager extends GridCacheManagerAdapter implements
cache.metrics0().onEvict();
if (recordable)
- cctx.events().addEvent(entry.partition(), entry.key(), cctx.nodeId(), (IgniteUuid)null, null,
+ cctx.events().addEvent(entry.partition(), entry.key(), cctx.nodeId(), null, null, null,
EVT_CACHE_ENTRY_EVICTED, null, false, oldVal, hasVal, null, null, null, false);
if (log.isDebugEnabled())
@@ -296,7 +296,7 @@ public class GridCacheEvictionManager extends GridCacheManagerAdapter implements
notifyPolicy(entry);
if (recordable)
- cctx.events().addEvent(entry.partition(), entry.key(), cctx.nodeId(), (IgniteUuid)null, null,
+ cctx.events().addEvent(entry.partition(), entry.key(), cctx.nodeId(), null, null, null,
EVT_CACHE_ENTRY_EVICTED, null, false, entry.rawGet(), entry.hasValue(), null, null, null,
false);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
index 95ce2b9..9124884 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMapEntry.java
@@ -1607,7 +1607,8 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
cctx.events().addEvent(partition(),
key,
evtNodeId,
- tx == null ? null : tx.xid(),
+ tx,
+ null,
newVer,
EVT_CACHE_OBJECT_PUT,
val,
@@ -1818,7 +1819,9 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
cctx.events().addEvent(partition(),
key,
evtNodeId,
- tx == null ? null : tx.xid(), newVer,
+ tx,
+ null,
+ newVer,
EVT_CACHE_OBJECT_REMOVED,
null,
false,
@@ -2152,7 +2155,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (transformCloClsName != null && cctx.events().isRecordable(EVT_CACHE_OBJECT_READ)) {
evtOld = cctx.unwrapTemporary(old);
- cctx.events().addEvent(partition(), key, cctx.localNodeId(), null,
+ cctx.events().addEvent(partition(), key, cctx.localNodeId(),null, null,
(GridCacheVersion)null, EVT_CACHE_OBJECT_READ, evtOld, evtOld != null || hadVal, evtOld,
evtOld != null || hadVal, subjId, transformCloClsName, taskName, keepBinary);
}
@@ -2161,7 +2164,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (evtOld == null)
evtOld = cctx.unwrapTemporary(old);
- cctx.events().addEvent(partition(), key, cctx.localNodeId(), null,
+ cctx.events().addEvent(partition(), key, cctx.localNodeId(), null, null,
(GridCacheVersion)null, EVT_CACHE_OBJECT_PUT, updated, updated != null, evtOld,
evtOld != null || hadVal, subjId, null, taskName, keepBinary);
}
@@ -2182,7 +2185,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
CacheObject evtOld = null;
if (transformCloClsName != null && cctx.events().isRecordable(EVT_CACHE_OBJECT_READ))
- cctx.events().addEvent(partition(), key, cctx.localNodeId(), null,
+ cctx.events().addEvent(partition(), key, cctx.localNodeId(), null, null,
(GridCacheVersion)null, EVT_CACHE_OBJECT_READ, evtOld, evtOld != null || hadVal, evtOld,
evtOld != null || hadVal, subjId, transformCloClsName, taskName, keepBinary);
@@ -2190,7 +2193,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
if (evtOld == null)
evtOld = cctx.unwrapTemporary(old);
- cctx.events().addEvent(partition(), key, cctx.localNodeId(), null, (GridCacheVersion)null,
+ cctx.events().addEvent(partition(), key, cctx.localNodeId(), null, null, (GridCacheVersion)null,
EVT_CACHE_OBJECT_REMOVED, null, false, evtOld, evtOld != null || hadVal, subjId, null,
taskName, keepBinary);
}
@@ -2427,6 +2430,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
key,
evtNodeId,
null,
+ null,
updateVer,
EVT_CACHE_OBJECT_READ,
evtOld, evtOld != null,
@@ -2454,6 +2458,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
key,
evtNodeId,
null,
+ null,
updateVer,
EVT_CACHE_OBJECT_PUT,
updateVal,
@@ -2483,6 +2488,7 @@ public abstract class GridCacheMapEntry extends GridMetadataAwareAdapter impleme
key,
evtNodeId,
null,
+ null,
updateVer,
EVT_CACHE_OBJECT_REMOVED,
null, false,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedTxRemoteAdapter.java
----------------------------------------------------------------------
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 6923080..c93e771 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
@@ -124,6 +124,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}.
*/
@@ -145,6 +149,7 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
* @param txSize Expected transaction size.
* @param subjId Subject ID.
* @param taskNameHash Task name hash code.
+ * @param txLbl Transaction label.
*/
public GridDistributedTxRemoteAdapter(
GridCacheSharedContext<?, ?> ctx,
@@ -159,7 +164,8 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
long timeout,
int txSize,
@Nullable UUID subjId,
- int taskNameHash
+ int taskNameHash,
+ String txLbl
) {
super(
ctx,
@@ -177,6 +183,7 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
taskNameHash);
this.invalidate = invalidate;
+ this.txLbl = txLbl;
commitVersion(commitVer);
@@ -992,6 +999,11 @@ public abstract class GridDistributedTxRemoteAdapter extends IgniteTxAdapter
}
/** {@inheritDoc} */
+ @Override public String label() {
+ return txLbl;
+ }
+
+ /** {@inheritDoc} */
@Override public String toString() {
return GridToStringBuilder.toString(GridDistributedTxRemoteAdapter.class, this, "super", super.toString());
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtCacheAdapter.java
----------------------------------------------------------------------
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 5628230..faada54 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 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @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 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
boolean recovery,
+ @Nullable String txLbl,
MvccSnapshot mvccSnapshot
) {
return getAllAsync0(keys,
@@ -814,6 +816,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
/*keep cache objects*/true,
recovery,
/*need version*/true,
+ txLbl,
mvccSnapshot);
}
@@ -828,6 +831,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @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 abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
boolean recovery,
+ @Nullable String txLbl,
MvccSnapshot mvccSnapshot
) {
GridDhtGetFuture<K, V> fut = new GridDhtGetFuture<>(ctx,
@@ -856,6 +861,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
skipVals,
recovery,
addReaders,
+ txLbl,
mvccSnapshot);
fut.init();
@@ -874,6 +880,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
* @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 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
@Nullable IgniteCacheExpiryPolicy expiry,
boolean skipVals,
boolean recovery,
+ String txLbl,
MvccSnapshot mvccSnapshot
) {
GridDhtGetSingleFuture fut = new GridDhtGetSingleFuture<>(
@@ -904,6 +912,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
expiry,
skipVals,
recovery,
+ txLbl,
mvccSnapshot);
fut.init();
@@ -933,6 +942,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
expiryPlc,
req.skipValues(),
req.recovery(),
+ req.txLabel(),
req.mvccSnapshot());
fut.listen(new CI1<IgniteInternalFuture<GridCacheEntryInfo>>() {
@@ -1037,6 +1047,7 @@ public abstract class GridDhtCacheAdapter<K, V> extends GridDistributedCacheAdap
expiryPlc,
req.skipValues(),
req.recovery(),
+ req.txLabel(),
req.mvccSnapshot());
fut.listen(new CI1<IgniteInternalFuture<Collection<GridCacheEntryInfo>>>() {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetFuture.java
----------------------------------------------------------------------
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 0bdc6b1..024e262 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
@@ -117,6 +117,9 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
/** */
private final boolean addReaders;
+ /** Transaction label. */
+ private final String txLbl;
+
/** */
private final MvccSnapshot mvccSnapshot;
@@ -131,6 +134,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
* @param taskNameHash Task name hash code.
* @param expiryPlc Expiry policy.
* @param skipVals Skip values flag.
+ * @param txLbl Transaction label.
* @param mvccSnapshot MVCC snapshot.
*/
public GridDhtGetFuture(
@@ -146,6 +150,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
boolean skipVals,
boolean recovery,
boolean addReaders,
+ @Nullable String txLbl,
MvccSnapshot mvccSnapshot
) {
super(CU.<GridCacheEntryInfo>collectionsReducer(keys.size()));
@@ -165,6 +170,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
this.skipVals = skipVals;
this.recovery = recovery;
this.addReaders = addReaders;
+ this.txLbl = txLbl;
this.mvccSnapshot = mvccSnapshot;
futId = IgniteUuid.randomUuid();
@@ -431,6 +437,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
expiryPlc,
skipVals,
recovery,
+ txLbl,
mvccSnapshot);
}
else {
@@ -455,6 +462,7 @@ public final class GridDhtGetFuture<K, V> extends GridCompoundIdentityFuture<Col
expiryPlc,
skipVals,
recovery,
+ txLbl,
mvccSnapshot);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtGetSingleFuture.java
----------------------------------------------------------------------
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 ee46168..88f6848 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
@@ -106,6 +106,9 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
/** Recovery context flag. */
private final boolean recovery;
+ /** Transaction label. */
+ private final String txLbl;
+
/** */
private final MvccSnapshot mvccSnapshot;
@@ -121,6 +124,7 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
* @param taskNameHash Task name hash code.
* @param expiryPlc Expiry policy.
* @param skipVals Skip values flag.
+ * @param txLbl Transaction label.
* @param mvccSnapshot Mvcc snapshot.
*/
public GridDhtGetSingleFuture(
@@ -136,6 +140,7 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
@Nullable IgniteCacheExpiryPolicy expiryPlc,
boolean skipVals,
boolean recovery,
+ @Nullable String txLbl,
@Nullable MvccSnapshot mvccSnapshot
) {
assert reader != null;
@@ -153,6 +158,7 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
this.expiryPlc = expiryPlc;
this.skipVals = skipVals;
this.recovery = recovery;
+ this.txLbl = txLbl;
this.mvccSnapshot = mvccSnapshot;
futId = IgniteUuid.randomUuid();
@@ -374,6 +380,7 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
expiryPlc,
skipVals,
recovery,
+ txLbl,
mvccSnapshot);
}
else {
@@ -400,6 +407,7 @@ public final class GridDhtGetSingleFuture<K, V> extends GridFutureAdapter<GridCa
expiryPlc,
skipVals,
recovery,
+ null,
mvccSnapshot);
fut0.listen(createGetFutureListener());
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
index 7ba0257..9361b30 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockFuture.java
@@ -945,7 +945,8 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
skipStore,
cctx.store().configured(),
keepBinary,
- cctx.deploymentEnabled());
+ cctx.deploymentEnabled(),
+ inTx() ? tx.label() : null);
try {
for (ListIterator<GridDhtCacheEntry> it = dhtMapping.listIterator(); it.hasNext(); ) {
@@ -1320,8 +1321,8 @@ public final class GridDhtLockFuture extends GridCacheCompoundIdentityFuture<Boo
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);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtLockRequest.java
----------------------------------------------------------------------
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 1ac5818..090ea68 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 class GridDhtLockRequest extends GridDistributedLockRequest {
* @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 class GridDhtLockRequest extends GridDistributedLockRequest {
boolean skipStore,
boolean storeUsed,
boolean keepBinary,
- boolean addDepInfo
+ boolean addDepInfo,
+ String txLbl
) {
super(cacheId,
nodeId,
@@ -179,6 +184,8 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
this.subjId = subjId;
this.taskNameHash = taskNameHash;
this.accessTtl = accessTtl;
+
+ this.txLbl = txLbl;
}
/**
@@ -309,6 +316,13 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
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 class GridDhtLockRequest extends GridDistributedLockRequest {
writer.incrementState();
+ case 30:
+ if (!writer.writeString("txLbl", txLbl))
+ return false;
+
+ writer.incrementState();
+
}
return true;
@@ -519,6 +539,14 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
reader.incrementState();
+ case 30:
+ txLbl = reader.readString("txLbl");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
}
return reader.afterMessageRead(GridDhtLockRequest.class);
@@ -531,7 +559,7 @@ public class GridDhtLockRequest extends GridDistributedLockRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 30;
+ return 31;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
----------------------------------------------------------------------
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 f3ca633..5437991 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
@@ -338,7 +338,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
req.txSize(),
req.subjectId(),
req.taskNameHash(),
- !req.skipStore() && req.storeUsed());
+ !req.skipStore() && req.storeUsed(),
+ req.txLabel());
tx = ctx.tm().onCreated(null, tx);
@@ -1120,7 +1121,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
req.txSize(),
null,
req.subjectId(),
- req.taskNameHash());
+ req.taskNameHash(),
+ req.txLabel());
if (req.syncCommit())
tx.syncMode(FULL_SYNC);
@@ -2151,7 +2153,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
-1,
null,
txSubjectId,
- txTaskNameHash);
+ txTaskNameHash,
+ null);
// if (req.syncCommit())
tx.syncMode(FULL_SYNC);
@@ -2278,7 +2281,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
-1,
req0.subjectId(),
req0.taskNameHash(),
- false);
+ false,
+ null);
tx.mvccSnapshot(new MvccSnapshotWithoutTxs(req0.coordinatorVersion(), req0.counter(),
MVCC_OP_COUNTER_NA, req0.cleanupVersion()));
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocal.java
----------------------------------------------------------------------
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 8550bdf..6bad093 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
@@ -96,6 +96,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}.
*/
@@ -118,6 +121,7 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
* @param storeEnabled Store enabled flag.
* @param txSize Expected transaction size.
* @param txNodes Transaction nodes mapping.
+ * @param lb Transaction label.
*/
public GridDhtTxLocal(
GridCacheSharedContext cctx,
@@ -141,7 +145,8 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
int txSize,
Map<UUID, Collection<UUID>> txNodes,
UUID subjId,
- int taskNameHash
+ int taskNameHash,
+ @Nullable String lb
) {
super(
cctx,
@@ -161,6 +166,8 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
subjId,
taskNameHash);
+ this.lb = lb;
+
assert nearNodeId != null;
assert nearFutId != null;
assert nearXidVer != null;
@@ -533,6 +540,11 @@ public class GridDhtTxLocal extends GridDhtTxLocalAdapter implements GridCacheMa
}
/** {@inheritDoc} */
+ @Nullable @Override public String label() {
+ return lb;
+ }
+
+ /** {@inheritDoc} */
@Override protected void clearPrepareFuture(GridDhtTxPrepareFuture fut) {
assert optimistic();
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareFuture.java
----------------------------------------------------------------------
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 6608b37..946b2a2 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
@@ -1894,8 +1894,8 @@ public final class GridDhtTxPrepareFuture extends GridCacheCompoundFuture<Ignite
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)
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxPrepareRequest.java
----------------------------------------------------------------------
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 30e8ceb..c86265a 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}.
*/
@@ -188,6 +192,8 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
nearNodeId = tx.nearNodeId();
skipCompletedVers = tx.xidVersion() == tx.nearXidVersion();
+
+ txLbl = tx.label();
}
/**
@@ -333,6 +339,13 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
}
/**
+ * @return Transaction label.
+ */
+ @Nullable public String txLabel() {
+ return txLbl;
+ }
+
+ /**
* {@inheritDoc}
*
* @param ctx
@@ -424,91 +437,97 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
switch (writer.state()) {
case 20:
- if (!writer.writeIgniteUuid("futId", futId))
+ if (!writer.writeCollection("counters", counters, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 21:
- if (!writer.writeBitSet("invalidateNearEntries", invalidateNearEntries))
+ if (!writer.writeIgniteUuid("futId", futId))
return false;
writer.incrementState();
case 22:
- if (!writer.writeInt("miniId", miniId))
+ if (!writer.writeBitSet("invalidateNearEntries", invalidateNearEntries))
return false;
writer.incrementState();
case 23:
- if (!writer.writeUuid("nearNodeId", nearNodeId))
+ if (!writer.writeInt("miniId", miniId))
return false;
writer.incrementState();
case 24:
- if (!writer.writeCollection("nearWrites", nearWrites, MessageCollectionItemType.MSG))
+ if (!writer.writeMessage("mvccSnapshot", mvccSnapshot))
return false;
writer.incrementState();
case 25:
- if (!writer.writeMessage("nearXidVer", nearXidVer))
+ if (!writer.writeUuid("nearNodeId", nearNodeId))
return false;
writer.incrementState();
case 26:
- if (!writer.writeCollection("ownedKeys", ownedKeys, MessageCollectionItemType.MSG))
+ if (!writer.writeCollection("nearWrites", nearWrites, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 27:
- if (!writer.writeCollection("ownedVals", ownedVals, MessageCollectionItemType.MSG))
+ if (!writer.writeMessage("nearXidVer", nearXidVer))
return false;
writer.incrementState();
case 28:
- if (!writer.writeBitSet("preloadKeys", preloadKeys))
+ if (!writer.writeCollection("ownedKeys", ownedKeys, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 29:
- if (!writer.writeBoolean("skipCompletedVers", skipCompletedVers))
+ if (!writer.writeCollection("ownedVals", ownedVals, MessageCollectionItemType.MSG))
return false;
writer.incrementState();
case 30:
- if (!writer.writeUuid("subjId", subjId))
+ if (!writer.writeBitSet("preloadKeys", preloadKeys))
return false;
writer.incrementState();
case 31:
- if (!writer.writeInt("taskNameHash", taskNameHash))
+ if (!writer.writeBoolean("skipCompletedVers", skipCompletedVers))
return false;
writer.incrementState();
case 32:
- if (!writer.writeMessage("topVer", topVer))
+ if (!writer.writeUuid("subjId", subjId))
return false;
writer.incrementState();
case 33:
- if (!writer.writeMessage("mvccSnapshot", mvccSnapshot))
+ if (!writer.writeInt("taskNameHash", taskNameHash))
return false;
writer.incrementState();
case 34:
- if (!writer.writeCollection("counters", counters, MessageCollectionItemType.MSG))
+ if (!writer.writeMessage("topVer", topVer))
+ return false;
+
+ writer.incrementState();
+
+ case 35:
+ if (!writer.writeString("txLbl", txLbl))
return false;
writer.incrementState();
@@ -530,7 +549,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
switch (reader.state()) {
case 20:
- futId = reader.readIgniteUuid("futId");
+ counters = reader.readCollection("counters", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
@@ -538,7 +557,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 21:
- invalidateNearEntries = reader.readBitSet("invalidateNearEntries");
+ futId = reader.readIgniteUuid("futId");
if (!reader.isLastRead())
return false;
@@ -546,7 +565,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 22:
- miniId = reader.readInt("miniId");
+ invalidateNearEntries = reader.readBitSet("invalidateNearEntries");
if (!reader.isLastRead())
return false;
@@ -554,7 +573,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 23:
- nearNodeId = reader.readUuid("nearNodeId");
+ miniId = reader.readInt("miniId");
if (!reader.isLastRead())
return false;
@@ -562,7 +581,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 24:
- nearWrites = reader.readCollection("nearWrites", MessageCollectionItemType.MSG);
+ mvccSnapshot = reader.readMessage("mvccSnapshot");
if (!reader.isLastRead())
return false;
@@ -570,7 +589,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 25:
- nearXidVer = reader.readMessage("nearXidVer");
+ nearNodeId = reader.readUuid("nearNodeId");
if (!reader.isLastRead())
return false;
@@ -578,7 +597,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 26:
- ownedKeys = reader.readCollection("ownedKeys", MessageCollectionItemType.MSG);
+ nearWrites = reader.readCollection("nearWrites", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
@@ -586,7 +605,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 27:
- ownedVals = reader.readCollection("ownedVals", MessageCollectionItemType.MSG);
+ nearXidVer = reader.readMessage("nearXidVer");
if (!reader.isLastRead())
return false;
@@ -594,7 +613,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 28:
- preloadKeys = reader.readBitSet("preloadKeys");
+ ownedKeys = reader.readCollection("ownedKeys", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
@@ -602,7 +621,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 29:
- skipCompletedVers = reader.readBoolean("skipCompletedVers");
+ ownedVals = reader.readCollection("ownedVals", MessageCollectionItemType.MSG);
if (!reader.isLastRead())
return false;
@@ -610,7 +629,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 30:
- subjId = reader.readUuid("subjId");
+ preloadKeys = reader.readBitSet("preloadKeys");
if (!reader.isLastRead())
return false;
@@ -618,7 +637,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 31:
- taskNameHash = reader.readInt("taskNameHash");
+ skipCompletedVers = reader.readBoolean("skipCompletedVers");
if (!reader.isLastRead())
return false;
@@ -626,7 +645,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 32:
- topVer = reader.readMessage("topVer");
+ subjId = reader.readUuid("subjId");
if (!reader.isLastRead())
return false;
@@ -634,7 +653,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 33:
- mvccSnapshot = reader.readMessage("mvccSnapshot");
+ taskNameHash = reader.readInt("taskNameHash");
if (!reader.isLastRead())
return false;
@@ -642,7 +661,15 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
reader.incrementState();
case 34:
- counters = reader.readCollection("counters", MessageCollectionItemType.MSG);
+ topVer = reader.readMessage("topVer");
+
+ if (!reader.isLastRead())
+ return false;
+
+ reader.incrementState();
+
+ case 35:
+ txLbl = reader.readString("txLbl");
if (!reader.isLastRead())
return false;
@@ -661,7 +688,7 @@ public class GridDhtTxPrepareRequest extends GridDistributedTxPrepareRequest {
/** {@inheritDoc} */
@Override public byte fieldsCount() {
- return 35;
+ return 36;
}
/** {@inheritDoc} */
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxRemote.java
----------------------------------------------------------------------
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 259c311..9518a01 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 class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
* @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 class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
@Nullable UUID subjId,
int taskNameHash,
boolean single,
- boolean storeWriteThrough) {
+ boolean storeWriteThrough,
+ @Nullable String txLbl) {
super(
ctx,
nodeId,
@@ -129,7 +131,8 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
timeout,
txSize,
subjId,
- taskNameHash
+ taskNameHash,
+ txLbl
);
assert nearNodeId != null;
@@ -169,6 +172,7 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
* @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 class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
int txSize,
@Nullable UUID subjId,
int taskNameHash,
- boolean storeWriteThrough) {
+ boolean storeWriteThrough,
+ @Nullable String txLbl) {
super(
ctx,
nodeId,
@@ -202,7 +207,8 @@ public class GridDhtTxRemote extends GridDistributedTxRemoteAdapter {
timeout,
txSize,
subjId,
- taskNameHash
+ taskNameHash,
+ txLbl
);
assert nearNodeId != null;
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedGetFuture.java
----------------------------------------------------------------------
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 f8a4ba4..8725e05 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
@@ -80,6 +80,9 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
/** Logger. */
private static IgniteLogger log;
+ /** Transaction label. */
+ private String txLbl;
+
/** */
protected final MvccSnapshot mvccSnapshot;
@@ -100,6 +103,7 @@ public class GridPartitionedGetFuture<K, V> 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(
@@ -115,6 +119,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
boolean skipVals,
boolean needVer,
boolean keepCacheObjects,
+ @Nullable String txLbl,
@Nullable MvccSnapshot mvccSnapshot
) {
super(cctx,
@@ -133,6 +138,8 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
this.mvccSnapshot = mvccSnapshot;
+ this.txLbl = txLbl;
+
if (log == null)
log = U.logger(cctx.kernalContext(), logRef, GridPartitionedGetFuture.class);
}
@@ -359,6 +366,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
expiryPlc,
skipVals,
recovery,
+ txLbl,
mvccSnapshot());
final Collection<Integer> invalidParts = fut.invalidPartitions();
@@ -417,6 +425,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
skipVals,
cctx.deploymentEnabled(),
recovery,
+ txLbl,
mvccSnapshot());
add(fut); // Append new future.
@@ -546,6 +555,7 @@ public class GridPartitionedGetFuture<K, V> extends CacheDistributedGetFutureAda
if (evt) {
cctx.events().readEvent(key,
null,
+ txLbl,
row.value(),
subjId,
taskName,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridPartitionedSingleGetFuture.java
----------------------------------------------------------------------
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 f28a969..ad716e6 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
@@ -132,6 +132,9 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
/** Post processing closure. */
private volatile BackupPostProcessingClosure postProcessingClos;
+ /** Transaction label. */
+ private String txLbl;
+
/**
* @param cctx Context.
* @param key Key.
@@ -145,6 +148,7 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
* @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.
*/
public GridPartitionedSingleGetFuture(
GridCacheContext cctx,
@@ -160,6 +164,7 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
boolean needVer,
boolean keepCacheObjects,
boolean recovery,
+ String txLbl,
@Nullable MvccSnapshot mvccSnapshot
) {
assert key != null;
@@ -190,6 +195,8 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
this.topVer = topVer;
this.mvccSnapshot = mvccSnapshot;
+ this.txLbl = txLbl;
+
futId = IgniteUuid.randomUuid();
if (log == null)
@@ -244,6 +251,7 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
expiryPlc,
skipVals,
recovery,
+ txLbl,
mvccSnapshot);
final Collection<Integer> invalidParts = fut.invalidPartitions();
@@ -315,6 +323,7 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
needVer,
cctx.deploymentEnabled(),
recovery,
+ txLbl,
mvccSnapshot);
try {
@@ -406,6 +415,7 @@ public class GridPartitionedSingleGetFuture extends GridCacheFutureAdapter<Objec
if (evt) {
cctx.events().readEvent(key,
null,
+ txLbl,
row.value(),
subjId,
taskName,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
index 686aace..cbb217c 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/atomic/GridDhtAtomicCache.java
@@ -1420,6 +1420,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
needVer,
false,
recovery,
+ null,
null);
fut.init();
@@ -1491,6 +1492,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
if (evt) {
ctx.events().readEvent(key,
null,
+ null,
row.value(),
subjId,
taskName,
@@ -1627,6 +1629,7 @@ public class GridDhtAtomicCache<K, V> extends GridDhtCacheAdapter<K, V> {
skipVals,
needVer,
false,
+ null,
null);
fut.init(topVer);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
index bbfb502..8061362 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedCache.java
@@ -267,6 +267,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
needVer,
/*keepCacheObjects*/false,
opCtx != null && opCtx.recovery(),
+ null,
mvccSnapshot);
fut.init();
@@ -364,7 +365,9 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
skipVals,
needVer,
false,
- mvccSnapshot);
+ null,
+ mvccSnapshot
+ );
if(mvccTracker != null){
final MvccQueryTracker mvccTracker0 = mvccTracker;
@@ -382,6 +385,47 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
}
/**
+ * @param keys Keys to load.
+ * @param readThrough Read through flag.
+ * @param forcePrimary Force get from primary node flag.
+ * @param topVer Topology version.
+ * @param subjId Subject ID.
+ * @param taskName Task name.
+ * @param deserializeBinary Deserialize binary flag.
+ * @param expiryPlc Expiry policy.
+ * @param skipVals Skip values flag.
+ * @param needVer Need version.
+ * @return Loaded values.
+ */
+ private IgniteInternalFuture<Map<K, V>> loadAsync(
+ @Nullable Collection<KeyCacheObject> keys,
+ boolean readThrough,
+ boolean forcePrimary,
+ AffinityTopologyVersion topVer,
+ @Nullable UUID subjId,
+ String taskName,
+ boolean deserializeBinary,
+ boolean recovery,
+ @Nullable IgniteCacheExpiryPolicy expiryPlc,
+ boolean skipVals,
+ boolean needVer,
+ @Nullable String txLbl) {
+ return loadAsync(keys,
+ readThrough,
+ forcePrimary,
+ topVer, subjId,
+ taskName,
+ deserializeBinary,
+ recovery,
+ expiryPlc,
+ skipVals,
+ needVer,
+ false,
+ txLbl,
+ null);
+ }
+
+ /**
* @param key Key to load.
* @param readThrough Read through flag.
* @param forcePrimary Force get from primary node flag.
@@ -393,6 +437,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
* @param skipVals Skip values flag.
* @param needVer If {@code true} returns values as tuples containing value and version.
* @param keepCacheObj Keep cache objects flag.
+ * @param txLbl Transaction label.
* @return Load future.
*/
public final IgniteInternalFuture<Object> loadAsync(
@@ -408,7 +453,8 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
boolean needVer,
boolean keepCacheObj,
boolean recovery,
- @Nullable MvccSnapshot mvccSnapshot
+ @Nullable MvccSnapshot mvccSnapshot,
+ @Nullable String txLbl
) {
GridPartitionedSingleGetFuture fut = new GridPartitionedSingleGetFuture(ctx,
ctx.toCacheKeyObject(key),
@@ -423,6 +469,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
needVer,
keepCacheObj,
recovery,
+ txLbl,
mvccSnapshot);
fut.init();
@@ -442,6 +489,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
* @param skipVals Skip values flag.
* @param needVer If {@code true} returns values as tuples containing value and version.
* @param keepCacheObj Keep cache objects flag.
+ * @param txLbl Transaction label.
* @param mvccSnapshot Mvcc snapshot.
* @return Load future.
*/
@@ -458,6 +506,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
boolean skipVals,
boolean needVer,
boolean keepCacheObj,
+ @Nullable String txLbl,
@Nullable MvccSnapshot mvccSnapshot
) {
assert mvccSnapshot == null || ctx.mvccEnabled();
@@ -507,6 +556,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
if (evt) {
ctx.events().readEvent(key,
null,
+ txLbl,
row.value(),
subjId,
taskName,
@@ -649,6 +699,7 @@ public class GridDhtColocatedCache<K, V> extends GridDhtTransactionalCacheAdapte
skipVals,
needVer,
keepCacheObj,
+ txLbl,
mvccSnapshot);
fut.init(topVer);
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
index f372769..6b20eb2 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java
@@ -1072,7 +1072,8 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF
keepBinary,
clientFirst,
false,
- cctx.deploymentEnabled());
+ cctx.deploymentEnabled(),
+ inTx() ? tx.label() : null);
mapping.request(req);
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
index b37acf3..323fe75 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtForceKeysFuture.java
@@ -552,8 +552,8 @@ public final class GridDhtForceKeysFuture<K, V> extends GridCompoundFuture<Objec
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);
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
index 332b09e..7ea943d 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/preloader/GridDhtPartitionDemander.java
@@ -933,8 +933,8 @@ public class GridDhtPartitionDemander {
cached.touch(topVer); // Start tracking.
if (cctx.events().isRecordable(EVT_CACHE_REBALANCE_OBJECT_LOADED) && !cached.isInternal())
- cctx.events().addEvent(cached.partition(), cached.key(), cctx.localNodeId(),
- (IgniteUuid)null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, entry.value(), true, null,
+ cctx.events().addEvent(cached.partition(), cached.key(), cctx.localNodeId(), null,
+ null, null, EVT_CACHE_REBALANCE_OBJECT_LOADED, entry.value(), true, null,
false, null, null, null, true);
}
else {
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/topology/GridDhtLocalPartition.java
----------------------------------------------------------------------
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 9151159..c1f3713 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
@@ -1089,7 +1089,8 @@ public class GridDhtLocalPartition extends GridCacheConcurrentMapImpl implements
hld.cctx.events().addEvent(cached.partition(),
cached.key(),
ctx.localNodeId(),
- (IgniteUuid)null,
+ null,
+ null,
null,
EVT_CACHE_REBALANCE_OBJECT_UNLOADED,
null,
http://git-wip-us.apache.org/repos/asf/ignite/blob/3c54f389/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
index 62a5dbd..0350e1a 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearGetFuture.java
@@ -319,6 +319,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
expiryPlc,
skipVals,
recovery,
+ null,
null); // TODO IGNITE-7371
final Collection<Integer> invalidParts = fut.invalidPartitions();
@@ -383,6 +384,7 @@ public final class GridNearGetFuture<K, V> extends CacheDistributedGetFutureAdap
skipVals,
cctx.deploymentEnabled(),
recovery,
+ null,
null); // TODO IGNITE-7371
add(fut); // Append new future.