This is an automated email from the ASF dual-hosted git repository.
vldpyatkov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite.git
The following commit(s) were added to refs/heads/master by this push:
new a3d3d8b2a75 IGNITE-28727 Create internal API method to lock cache
entry with specific version (#13264)
a3d3d8b2a75 is described below
commit a3d3d8b2a75a505eff26aab80db49c09c786467e
Author: Vladislav Pyatkov <[email protected]>
AuthorDate: Tue Jun 30 09:29:38 2026 +0300
IGNITE-28727 Create internal API method to lock cache entry with specific
version (#13264)
---
.../processors/cache/GridCacheAdapter.java | 196 ++++++-
.../processors/cache/GridCacheProxyImpl.java | 52 ++
.../internal/processors/cache/GridCacheUtils.java | 11 +
.../processors/cache/IgniteInternalCache.java | 59 ++
.../distributed/GridDistributedCacheAdapter.java | 8 +-
.../cache/distributed/dht/GridDhtLockFuture.java | 65 ++-
.../dht/GridDhtTransactionalCacheAdapter.java | 63 ++-
.../distributed/dht/GridDhtTxLocalAdapter.java | 47 +-
.../distributed/dht/atomic/GridDhtAtomicCache.java | 1 +
.../dht/colocated/GridDhtColocatedCache.java | 18 +-
.../dht/colocated/GridDhtColocatedLockFuture.java | 62 ++-
.../distributed/near/GridNearAtomicCache.java | 1 +
.../cache/distributed/near/GridNearLockFuture.java | 82 ++-
.../distributed/near/GridNearLockRequest.java | 16 +-
.../distributed/near/GridNearLockResponse.java | 18 +
.../near/GridNearPessimisticTxPrepareFuture.java | 3 +-
.../near/GridNearTransactionalCache.java | 2 +
.../cache/distributed/near/GridNearTxLocal.java | 55 +-
.../cache/transactions/IgniteTxLocalAdapter.java | 56 +-
.../cache/transactions/IgniteTxManager.java | 5 +-
...CacheOperationContextTransactionalLockTest.java | 141 +++++
.../CacheVersionedEntryTransactionalLockTest.java | 514 ++++++++++++++++++
.../processors/cache/LockTxEntryOneNodeTest.java | 602 +++++++++++++++++++++
.../ignite/testsuites/IgniteCacheTestSuite4.java | 7 +
24 files changed, 1998 insertions(+), 86 deletions(-)
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 f1e23417728..2fe29746e7a 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
@@ -98,6 +98,7 @@ import
org.apache.ignite.internal.processors.cache.distributed.near.consistency.
import org.apache.ignite.internal.processors.cache.dr.GridCacheDrInfo;
import org.apache.ignite.internal.processors.cache.persistence.CacheDataRow;
import
org.apache.ignite.internal.processors.cache.transactions.IgniteInternalTx;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
import
org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalAdapter;
import
org.apache.ignite.internal.processors.cache.transactions.IgniteTxLocalEx;
@@ -159,6 +160,7 @@ import org.jetbrains.annotations.Nullable;
import static
org.apache.ignite.IgniteSystemProperties.IGNITE_CACHE_RETRIES_COUNT;
import static org.apache.ignite.internal.GridClosureCallMode.BROADCAST;
+import static
org.apache.ignite.internal.processors.cache.GridCacheOperation.READ;
import static
org.apache.ignite.internal.processors.cache.distributed.dht.topology.GridDhtPartitionState.OWNING;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_LOAD;
import static org.apache.ignite.internal.processors.dr.GridDrType.DR_NONE;
@@ -538,7 +540,8 @@ public abstract class GridCacheAdapter<K, V> implements
IgniteInternalCache<K, V
/**
* @param keys Keys to lock.
- * @param timeout Lock timeout.
+ * @param timeout Transaction timeout.
+ * @param waitTimeout Lock wait timeout.
* @param tx Transaction.
* @param isRead {@code True} for read operations.
* @param retval Flag to return value.
@@ -551,6 +554,7 @@ public abstract class GridCacheAdapter<K, V> implements
IgniteInternalCache<K, V
public abstract IgniteInternalFuture<Boolean> txLockAsync(
Collection<KeyCacheObject> keys,
long timeout,
+ long waitTimeout,
IgniteTxLocalEx tx,
boolean isRead,
boolean retval,
@@ -3028,6 +3032,196 @@ public abstract class GridCacheAdapter<K, V> implements
IgniteInternalCache<K, V
}
}
+ /** {@inheritDoc} */
+ @Override public boolean lockTxEntry(CacheEntry<K, V> entry, long
waitTimeout) throws IgniteCheckedException {
+ A.notNull(entry, "entry");
+
+ return lockTxEntryAsync(entry, waitTimeout).get();
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean lockTxEntries(Collection<CacheEntry<K, V>>
entries, long waitTimeout)
+ throws IgniteCheckedException {
+ A.notNull(entries, "entries");
+
+ return lockTxEntriesAsync(entries, waitTimeout).get();
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<Boolean>
lockTxEntryAsync(CacheEntry<K, V> entry, long waitTimeout) {
+ A.notNull(entry, "entry");
+
+ return lockTxEntriesAsync(Collections.singleton(entry), waitTimeout);
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<Boolean> lockTxEntriesAsync(
+ Collection<CacheEntry<K, V>> entries,
+ long waitTimeout
+ ) {
+ A.notNull(entries, "entries");
+
+ GridNearTxLocal tx = tx();
+
+ if (tx == null)
+ return new GridFinishedFuture<>(
+ new IgniteCheckedException("Failed to acquire transactional
lock without transaction."));
+
+ if (!tx.pessimistic())
+ return new GridFinishedFuture<>(
+ new IgniteCheckedException("Failed to acquire transactional
lock in optimistic transaction."));
+
+ // Wait for previous per-transaction async operations to finish.
+ tx.txState().awaitLastFuture();
+
+ if (!tx.init())
+ return new GridFinishedFuture<>(new
IgniteTxRollbackCheckedException(
+ "Failed to acquire transactional lock because transaction has
been completed: " + tx));
+
+ if (entries.isEmpty())
+ return new GridFinishedFuture<>(true);
+
+ try {
+ tx.addActiveCache(ctx, false);
+ }
+ catch (IgniteCheckedException e) {
+ return new GridFinishedFuture<>(e);
+ }
+
+ Collection<KeyCacheObject> keys = new ArrayList<>(entries.size());
+ List<IgniteTxEntry> txEntries = new ArrayList<>(entries.size());
+ List<GridCacheVersion> expVers = new ArrayList<>(entries.size());
+ Set<IgniteTxKey> txKeys = new HashSet<>(entries.size());
+
+ CacheOperationContext opCtx = ctx.operationContextPerCall();
+
+ for (CacheEntry<K, V> entry : entries) {
+ A.notNull(entry, "entry");
+
+ KeyCacheObject key = ctx.toCacheKeyObject(entry.getKey());
+ IgniteTxKey txKey = ctx.txKey(key);
+
+ if (!txKeys.add(txKey))
+ continue;
+
+ IgniteTxEntry lockedTxEntry = tx.entry(txKey);
+
+ if (lockedTxEntry != null && (lockedTxEntry.op() != READ ||
lockedTxEntry.locked()))
+ continue;
+
+ if (!(entry.version() instanceof GridCacheVersion)) {
+ tx.removeAndUnlockTxEntries(txEntries);
+
+ return new GridFinishedFuture<>(new
IgniteCheckedException("Failed to acquire transactional lock for entry with " +
+ "unsupported version type [entry=" + entry + ", version="
+ entry.version() + ']'));
+ }
+
+ CacheObject val = ctx.toCacheObject(entry.getValue());
+ GridCacheEntryEx entryEx = ctx.isColocated() ?
ctx.colocated().entryExx(key, tx.topologyVersion(), true) : entryEx(key);
+
+ IgniteTxEntry txEntry = tx.addEntry(
+ READ,
+ val,
+ null,
+ null,
+ entryEx,
+ null,
+ null,
+ true,
+ -1L,
+ -1L,
+ null,
+ opCtx != null && opCtx.skipStore(),
+ opCtx != null && opCtx.skipReadThrough(),
+ opCtx != null && opCtx.keepBinaryInInterceptor(),
+ opCtx != null && opCtx.isKeepBinary(),
+ CU.isNearEnabled(ctx)
+ );
+
+ keys.add(key);
+ txEntries.add(txEntry);
+ expVers.add((GridCacheVersion)entry.version());
+ }
+
+ if (keys.isEmpty())
+ return new GridFinishedFuture<>(true);
+
+ // Acquire transactional lock future from concrete cache
implementation. Use txLockAsync which
+ // delegates to cache-specific lockAllAsync implementations for
distributed caches.
+ long timeout = tx.remainingTime();
+
+ IgniteInternalFuture<Boolean> lockFut = txLockAsync(keys,
+ timeout,
+ waitTimeout,
+ tx,
+ /*isRead*/true,
+ /*retval*/false,
+ tx.isolation(),
+ /*invalidate*/false,
+ /*createTtl*/0L,
+ /*accessTtl*/0L);
+
+ IgniteInternalFuture<Boolean> res = new GridEmbeddedFuture<>(
+ lockFut,
+ (locked, ex) -> {
+ if (ex != null)
+ return new GridFinishedFuture<>(ex);
+
+ if (!locked) {
+ tx.removeAndUnlockTxEntries(txEntries);
+
+ return new GridFinishedFuture<>(false);
+ }
+
+ try {
+ for (int i = 0; i < txEntries.size(); i++) {
+ GridCacheEntryEx cached = txEntries.get(i).cached();
+ EntryGetResult getRes = cached.innerGetVersioned(
+ null,
+ tx,
+ /*update-metrics*/false,
+ /*event*/false,
+ null,
+ tx.resolveTaskName(),
+ null,
+ false,
+ null);
+
+ if (getRes == null ||
!expVers.get(i).equals(getRes.version())) {
+ tx.removeAndUnlockTxEntries(txEntries);
+
+ return new GridFinishedFuture<>(false);
+ }
+ }
+
+ return new GridFinishedFuture<>(true);
+ }
+ catch (IgniteCheckedException | GridCacheEntryRemovedException
e) {
+ tx.removeAndUnlockTxEntries(txEntries);
+
+ return new GridFinishedFuture<>(e);
+ }
+ }
+ );
+
+ // Register this future in transaction's async-holder so that
subsequent operations
+ // that call tx.txState().awaitLastFuture() will wait for it.
+ GridCacheAdapter.FutureHolder holder = tx.txState().lastAsyncFuture();
+
+ if (holder != null) {
+ holder.lock();
+
+ try {
+ holder.saveFuture(res);
+ }
+ finally {
+ holder.unlock();
+ }
+ }
+
+ return res;
+ }
+
/** {@inheritDoc} */
@Override public boolean isLockedByThread(K key) {
A.notNull(key, "key");
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
index 0e335e063f5..21c1fa361fe 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProxyImpl.java
@@ -1311,6 +1311,58 @@ public class GridCacheProxyImpl<K, V> implements
IgniteInternalCache<K, V>, Exte
}
}
+ /** {@inheritDoc} */
+ @Override public boolean lockTxEntry(CacheEntry<K, V> entry, long
waitTimeout) throws IgniteCheckedException {
+ CacheOperationContext prev = gate.enter(opCtx);
+
+ try {
+ return delegate.lockTxEntry(entry, waitTimeout);
+ }
+ finally {
+ gate.leave(prev);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<Boolean>
lockTxEntryAsync(CacheEntry<K, V> entry, long waitTimeout) {
+ CacheOperationContext prev = gate.enter(opCtx);
+
+ try {
+ return delegate.lockTxEntryAsync(entry, waitTimeout);
+ }
+ finally {
+ gate.leave(prev);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public boolean lockTxEntries(Collection<CacheEntry<K, V>>
entries, long waitTimeout)
+ throws IgniteCheckedException {
+ CacheOperationContext prev = gate.enter(opCtx);
+
+ try {
+ return delegate.lockTxEntries(entries, waitTimeout);
+ }
+ finally {
+ gate.leave(prev);
+ }
+ }
+
+ /** {@inheritDoc} */
+ @Override public IgniteInternalFuture<Boolean> lockTxEntriesAsync(
+ Collection<CacheEntry<K, V>> entries,
+ long waitTimeout
+ ) {
+ CacheOperationContext prev = gate.enter(opCtx);
+
+ try {
+ return delegate.lockTxEntriesAsync(entries, waitTimeout);
+ }
+ finally {
+ gate.leave(prev);
+ }
+ }
+
/** {@inheritDoc} */
@Override public boolean isLockedByThread(K key) {
CacheOperationContext prev = gate.enter(opCtx);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
index 94377623724..6bd9c78267c 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheUtils.java
@@ -178,6 +178,17 @@ public class GridCacheUtils {
return cheatCacheId != 0 && id == cheatCacheId;
}
+ /**
+ * Checks whether the separate lock wait timeout expires before the
transaction timeout.
+ *
+ * @param waitTimeout Lock wait timeout. {@code 0} means that there is no
separate lock wait timeout.
+ * @param timeout Transaction timeout. {@code 0} means that the
transaction timeout is infinite.
+ * @return {@code True} if the separate lock wait timeout expires before
the transaction timeout.
+ */
+ public static boolean isWaitTimeoutExpiresFirst(long waitTimeout, long
timeout) {
+ return timeout >= 0 && waitTimeout != 0 && (timeout == 0 || timeout >
waitTimeout);
+ }
+
/** System cache name. */
public static final String UTILITY_CACHE_NAME = "ignite-sys-cache";
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
index 0dd481067fe..f98849c6761 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/IgniteInternalCache.java
@@ -1345,6 +1345,65 @@ public interface IgniteInternalCache<K, V> extends
Iterable<Cache.Entry<K, V>> {
*/
public boolean isLocked(K key);
+ /**
+ * Acquires a transactional lock for the cached object represented by the
given entry if the current cached version
+ * matches the entry version. This method works only in a {@link
TransactionConcurrency#PESSIMISTIC} transaction.
+ *
+ * @param entry Entry whose key, value and version should be used.
+ * @param waitTimeout Timeout in milliseconds to wait for lock to be
acquired
+ * ({@code 0} to use the transaction timeout, {@code -1} for
immediate failure if
+ * lock cannot be acquired immediately).
+ * @return {@code True} if lock was acquired with the same entry version.
+ * @throws IgniteCheckedException If lock acquisition resulted in an error.
+ * @throws NullPointerException If entry is {@code null}.
+ */
+ public boolean lockTxEntry(CacheEntry<K, V> entry, long waitTimeout)
throws IgniteCheckedException;
+
+ /**
+ * Acquires transactional locks for the cached objects represented by the
given entries if all current cached
+ * versions match the corresponding entry versions. This method works only
in a
+ * {@link TransactionConcurrency#PESSIMISTIC} transaction.
+ *
+ * @param entries Entries whose keys, values and versions should be used.
+ * @param waitTimeout Timeout in milliseconds to wait for locks to be
acquired
+ * ({@code 0} to use the transaction timeout, {@code -1} for
immediate failure if
+ * locks cannot be acquired immediately).
+ * @return {@code True} if all locks were acquired with the same entry
versions.
+ * @throws IgniteCheckedException If lock acquisition resulted in an error.
+ * @throws NullPointerException If entries is {@code null}.
+ */
+ public boolean lockTxEntries(Collection<CacheEntry<K, V>> entries, long
waitTimeout) throws IgniteCheckedException;
+
+ /**
+ * Asynchronously acquires a transactional lock for the cached object
represented by the given entry if the current
+ * cached version matches the entry version. This method works only in a
+ * {@link TransactionConcurrency#PESSIMISTIC} transaction.
+ *
+ * @param entry Entry whose key, value and version should be used.
+ * @param waitTimeout Timeout in milliseconds to wait for lock to be
acquired
+ * ({@code 0} to use the transaction timeout, {@code -1} for
immediate failure if
+ * lock cannot be acquired immediately).
+ * @return Future that resolves to {@code true} if the lock was acquired
and the versions matched, or to
+ * {@code false} otherwise.
+ * @throws NullPointerException If entry is {@code null}.
+ */
+ public IgniteInternalFuture<Boolean> lockTxEntryAsync(CacheEntry<K, V>
entry, long waitTimeout);
+
+ /**
+ * Asynchronously acquires transactional locks for the cached objects
represented by the given entries if all
+ * current cached versions match the corresponding entry versions. This
method works only in a
+ * {@link TransactionConcurrency#PESSIMISTIC} transaction.
+ *
+ * @param entries Entries whose keys, values and versions should be used.
+ * @param waitTimeout Timeout in milliseconds to wait for locks to be
acquired
+ * ({@code 0} to use the transaction timeout, {@code -1} for
immediate failure if
+ * locks cannot be acquired immediately).
+ * @return Future that resolves to {@code true} if all locks were acquired
and all versions matched, or to
+ * {@code false} otherwise.
+ * @throws NullPointerException If entries is {@code null}.
+ */
+ public IgniteInternalFuture<Boolean>
lockTxEntriesAsync(Collection<CacheEntry<K, V>> entries, long waitTimeout);
+
/**
* Checks if current thread owns a lock on this key.
* <p>
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
index 8d330906dd4..7586312d1bd 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/GridDistributedCacheAdapter.java
@@ -102,6 +102,7 @@ public abstract class GridDistributedCacheAdapter<K, V>
extends GridCacheAdapter
@Override public IgniteInternalFuture<Boolean> txLockAsync(
Collection<KeyCacheObject> keys,
long timeout,
+ long waitTimeout,
IgniteTxLocalEx tx,
boolean isRead,
boolean retval,
@@ -112,7 +113,7 @@ public abstract class GridDistributedCacheAdapter<K, V>
extends GridCacheAdapter
) {
assert tx != null;
- return lockAllAsync(keys, timeout, tx, isInvalidate, isRead, retval,
isolation, createTtl, accessTtl);
+ return lockAllAsync(keys, timeout, waitTimeout, tx, isInvalidate,
isRead, retval, isolation, createTtl, accessTtl);
}
/** {@inheritDoc} */
@@ -121,6 +122,7 @@ public abstract class GridDistributedCacheAdapter<K, V>
extends GridCacheAdapter
// Return value flag is true because we choose to bring values for
explicit locks.
return lockAllAsync(ctx.cacheKeysView(keys),
+ timeout,
timeout,
tx,
false,
@@ -133,7 +135,8 @@ public abstract class GridDistributedCacheAdapter<K, V>
extends GridCacheAdapter
/**
* @param keys Keys to lock.
- * @param timeout Timeout.
+ * @param timeout Transaction timeout.
+ * @param waitTimeout Lock wait timeout.
* @param tx Transaction
* @param isInvalidate Invalidation flag.
* @param isRead Indicates whether value is read or written.
@@ -145,6 +148,7 @@ public abstract class GridDistributedCacheAdapter<K, V>
extends GridCacheAdapter
*/
protected abstract IgniteInternalFuture<Boolean>
lockAllAsync(Collection<KeyCacheObject> keys,
long timeout,
+ long waitTimeout,
@Nullable IgniteTxLocalEx tx,
boolean isInvalidate,
boolean isRead,
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 fef43785bef..0f2e50892fb 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
@@ -154,9 +154,12 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
@GridToStringExclude
private LockTimeoutObject timeoutObj;
- /** Lock timeout. */
+ /** Transaction timeout. */
private final long timeout;
+ /** Lock wait timeout. */
+ private final long waitTimeout;
+
/** Transaction. */
private final GridDhtTxLocalAdapter tx;
@@ -201,7 +204,8 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
* @param cnt Number of keys to lock.
* @param read Read flag.
* @param needReturnVal Need return value flag.
- * @param timeout Lock acquisition timeout.
+ * @param timeout Transaction timeout.
+ * @param waitTimeout Lock wait timeout.
* @param tx Transaction.
* @param threadId Thread ID.
* @param accessTtl TTL for read operation.
@@ -219,6 +223,7 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
boolean read,
boolean needReturnVal,
long timeout,
+ long waitTimeout,
GridDhtTxLocalAdapter tx,
long threadId,
long createTtl,
@@ -241,6 +246,7 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
this.read = read;
this.needReturnVal = needReturnVal;
this.timeout = timeout;
+ this.waitTimeout = waitTimeout;
this.tx = tx;
this.createTtl = createTtl;
this.accessTtl = accessTtl;
@@ -435,18 +441,21 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
threadId,
lockVer,
null,
- timeout,
+ lockTimeout(),
/*reenter*/false,
inTx(),
implicitSingle(),
false
);
- if (c == null && timeout < 0) {
+ if (c == null && lockTimeout() < 0) {
if (log.isDebugEnabled())
log.debug("Failed to acquire lock with negative timeout: " +
entry);
- onFailed();
+ if (CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout))
+ onComplete(false, false, false, false);
+ else
+ onFailed();
return null;
}
@@ -631,10 +640,13 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
try {
CacheLockCandidates owners = entry.readyLock(lockVer);
- if (timeout < 0) {
+ if (lockTimeout() < 0) {
if (owners == null || !owners.hasCandidate(lockVer)) {
// We did not send any requests yet.
- onFailed();
+ if (CU.isWaitTimeoutExpiresFirst(waitTimeout,
timeout))
+ onComplete(false, false, false, false);
+ else
+ onFailed();
return;
}
@@ -749,6 +761,9 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
this.err = err;
}
+ if (!success && err == null &&
CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout))
+ return onComplete(false, false, false, false);
+
return onComplete(success, err instanceof NodeStoppingException,
true);
}
}
@@ -762,13 +777,26 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
* @return {@code True} if complete by this operation.
*/
private synchronized boolean onComplete(boolean success, boolean stopping,
boolean unlock) {
+ return onComplete(success, stopping, unlock, !success);
+ }
+
+ /**
+ * Completeness callback.
+ *
+ * @param success {@code True} if lock was acquired.
+ * @param stopping {@code True} if node is stopping.
+ * @param unlock {@code True} if locks should be released.
+ * @param rollback {@code True} if should rollback tx on failure.
+ * @return {@code True} if complete by this operation.
+ */
+ private synchronized boolean onComplete(boolean success, boolean stopping,
boolean unlock, boolean rollback) {
if (log.isDebugEnabled())
log.debug("Received onComplete(..) callback [success=" + success +
", fut=" + this + ']');
if (isDone())
return false;
- if (!success && !stopping && unlock)
+ if (!success && !stopping && unlock && rollback)
undoLocks(true);
boolean set = false;
@@ -778,7 +806,7 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
set = cctx.tm().setTxTopologyHint(tx.topologyVersionSnapshot());
- if (success)
+ if (!rollback)
tx.clearLockFuture(this);
}
@@ -821,7 +849,7 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
readyLocks();
- if (timeout > 0 && !isDone()) { // Prevent memory leak if future
is completed by call to readyLocks.
+ if (lockTimeout() > 0 && !isDone()) { // Prevent memory leak if
future is completed by call to readyLocks.
timeoutObj = new LockTimeoutObject();
cctx.time().addTimeoutObject(timeoutObj);
@@ -1165,6 +1193,13 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
}
}
+ /**
+ * @return Timeout value for this lock future.
+ */
+ private long lockTimeout() {
+ return CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout) ?
waitTimeout : timeout;
+ }
+
/**
* Lock request timeout object.
*/
@@ -1173,7 +1208,7 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
* Default constructor.
*/
LockTimeoutObject() {
- super(timeout);
+ super(lockTimeout());
}
/** {@inheritDoc} */
@@ -1198,9 +1233,13 @@ public final class GridDhtLockFuture extends
GridCacheCompoundIdentityFuture<Boo
clear();
}
- boolean releaseLocks = !(inTx() &&
cctx.tm().deadlockDetectionEnabled());
+ if (CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout))
+ onComplete(false, false, false, false);
+ else {
+ boolean releaseLocks = !(inTx() &&
cctx.tm().deadlockDetectionEnabled());
- onComplete(false, false, releaseLocks);
+ onComplete(false, false, releaseLocks);
+ }
}
/** {@inheritDoc} */
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTransactionalCacheAdapter.java
index 56cc6e1e0bb..735da233ddf 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
@@ -721,6 +721,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
@Override public IgniteInternalFuture<Boolean> lockAllAsync(
@Nullable Collection<KeyCacheObject> keys,
long timeout,
+ long waitTimeout,
IgniteTxLocalEx txx,
boolean isInvalidate,
boolean isRead,
@@ -733,6 +734,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
return lockAllAsyncInternal(
keys,
timeout,
+ waitTimeout,
txx,
isInvalidate,
isRead,
@@ -750,7 +752,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
* Acquires locks in partitioned cache.
*
* @param keys Keys to lock.
- * @param timeout Lock timeout.
+ * @param timeout Transaction timeout.
+ * @param waitTimeout Lock wait timeout.
* @param txx Transaction.
* @param isInvalidate Invalidate flag.
* @param isRead Read flag.
@@ -765,6 +768,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
*/
public GridDhtFuture<Boolean> lockAllAsyncInternal(@Nullable
Collection<KeyCacheObject> keys,
long timeout,
+ long waitTimeout,
IgniteTxLocalEx txx,
boolean isInvalidate,
boolean isRead,
@@ -792,6 +796,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
isRead,
retval,
timeout,
+ waitTimeout,
tx,
tx.threadId(),
createTtl,
@@ -976,6 +981,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
req.txRead(),
req.needReturnValue(),
req.timeout(),
+ req.waitTimeout(),
tx,
req.threadId(),
req.createTtl(),
@@ -1058,6 +1064,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
req.skipReadThrough(),
req.keepBinaryInInterceptor(),
req.keepBinary(),
+ req.waitTimeout(),
req.nearCache());
final GridDhtTxLocal t = tx;
@@ -1071,7 +1078,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
e = U.unwrap(e);
// Transaction can be emptied by asynchronous
rollback.
- assert e != null || !t.empty();
+ boolean lockAcquired = e == null && o != null &&
o.success() && !t.empty();
// Create response while holding locks.
final GridNearLockResponse resp =
createLockReply(nearNode,
@@ -1079,7 +1086,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
req,
t,
t.xidVersion(),
- e);
+ e,
+ lockAcquired);
assert !t.implicit() : t;
assert !t.onePhaseCommit() : t;
@@ -1104,15 +1112,18 @@ public abstract class
GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
@Override public GridNearLockResponse apply(Boolean b,
Exception e) {
if (e != null)
e = U.unwrap(e);
- else if (!b)
+ else if (!b &&
!CU.isWaitTimeoutExpiresFirst(req.waitTimeout(), req.timeout()))
e = new
GridCacheLockTimeoutException(req.version());
+ boolean lockAcquired = e != null || b;
+
GridNearLockResponse res =
createLockReply(nearNode,
entries,
req,
null,
mappedVer,
- e);
+ e,
+ lockAcquired);
sendLockReply(nearNode, null, req, res);
@@ -1140,7 +1151,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
req,
tx,
tx != null ? tx.xidVersion() : req.version(),
- e);
+ e,
+ false);
sendLockReply(nearNode, null, req, res);
}
@@ -1197,6 +1209,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
* @param tx Transaction.
* @param mappedVer Mapped version.
* @param err Error.
+ * @param lockAcquired {@code True} if requested locks were acquired.
* @return Response.
*/
private GridNearLockResponse createLockReply(
@@ -1205,7 +1218,8 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
GridNearLockRequest req,
@Nullable GridDhtTxLocalAdapter tx,
GridCacheVersion mappedVer,
- Throwable err) {
+ Throwable err,
+ boolean lockAcquired) {
assert mappedVer != null;
assert tx == null || tx.xidVersion().equals(mappedVer);
@@ -1227,9 +1241,14 @@ public abstract class
GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
clienRemapVer,
clienRemapVer != null);
+ res.lockAcquired(lockAcquired);
+
if (err == null) {
res.pending(localDhtPendingVersions(entries, mappedVer));
+ if (!lockAcquired)
+ return res;
+
// We have to add completed versions for cases when nearLocal
and remote transactions
// execute concurrently.
IgnitePair<Collection<GridCacheVersion>> versPair =
ctx.tm().versions(req.version());
@@ -1251,6 +1270,21 @@ public abstract class
GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
GridCacheVersion ver = e.version();
+ boolean ownsLock = e.lockedBy(mappedVer) ||
+ ctx.mvcc().isRemoved(e.context(),
mappedVer);
+
+ if (!ownsLock &&
CU.isWaitTimeoutExpiresFirst(req.waitTimeout(), req.timeout())) {
+ res.lockAcquired(false);
+
+ return res;
+ }
+
+ assert ownsLock || tx != null &&
tx.isRollbackOnly() :
+ "Entry does not own lock for tx
[locNodeId=" + ctx.localNodeId() +
+ ", entry=" + e +
+ ", mappedVer=" + mappedVer + ", ver="
+ ver +
+ ", tx=" + CU.txString(tx) + ", req=" +
req + ']';
+
boolean ret = req.returnValue(i) || dhtVer ==
null || !dhtVer.equals(ver);
CacheObject val = null;
@@ -1268,14 +1302,6 @@ public abstract class
GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
req.keepBinary());
}
- assert e.lockedBy(mappedVer) ||
- ctx.mvcc().isRemoved(e.context(),
mappedVer) ||
- tx != null && tx.isRollbackOnly() :
- "Entry does not own lock for tx
[locNodeId=" + ctx.localNodeId() +
- ", entry=" + e +
- ", mappedVer=" + mappedVer + ", ver="
+ ver +
- ", tx=" + CU.txString(tx) + ", req=" +
req + ']';
-
boolean filterPassed = false;
if (tx != null && tx.onePhaseCommit()) {
@@ -1631,11 +1657,14 @@ public abstract class
GridDhtTransactionalCacheAdapter<K, V> extends GridDhtCach
GridCacheMvccCandidate cand = null;
if (dhtVer == null) {
- cand = entry.localCandidateByNearVersion(ver, true);
+ cand = entry.localCandidateByNearVersion(ver,
!forSavepoint);
if (cand != null)
dhtVer = cand.version();
else {
+ if (forSavepoint)
+ break;
+
if (log.isDebugEnabled())
log.debug("Failed to locate lock candidate
based on dht or near versions [nodeId=" +
nodeId + ", ver=" + ver + ", unmap=" +
unmap + ", keys=" + keys + ']');
@@ -1669,7 +1698,7 @@ public abstract class GridDhtTransactionalCacheAdapter<K,
V> extends GridDhtCach
// Note that we don't reorder completed versions here,
// as there is no point to reorder relative to the version
// we are about to remove.
- if (entry.removeLock(dhtVer)) {
+ if ((forSavepoint && cand == null) ||
entry.removeLock(dhtVer)) {
if (forSavepoint)
clearTxEntry(dhtVer, key);
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
index 625d59bb279..4c7e9313f5e 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/GridDhtTxLocalAdapter.java
@@ -580,6 +580,7 @@ public abstract class GridDhtTxLocalAdapter extends
IgniteTxLocalAdapter {
boolean skipReadThrough,
boolean keepBinaryInInterceptor,
boolean keepBinary,
+ long waitTimeout,
boolean nearCache
) {
try {
@@ -698,7 +699,8 @@ public abstract class GridDhtTxLocalAdapter extends
IgniteTxLocalAdapter {
skipStore,
skipReadThrough,
keepBinaryInInterceptor,
- keepBinary);
+ keepBinary,
+ waitTimeout);
}
catch (IgniteCheckedException e) {
setRollbackOnly();
@@ -731,7 +733,8 @@ public abstract class GridDhtTxLocalAdapter extends
IgniteTxLocalAdapter {
boolean skipStore,
boolean skipReadThrough,
boolean keepBinaryInInterceptor,
- boolean keepBinary) {
+ boolean keepBinary,
+ long waitTimeout) {
if (log.isDebugEnabled())
log.debug("Before acquiring transaction lock on keys [keys=" +
passedKeys + ']');
@@ -753,6 +756,7 @@ public abstract class GridDhtTxLocalAdapter extends
IgniteTxLocalAdapter {
IgniteInternalFuture<Boolean> fut =
dhtCache.lockAllAsyncInternal(passedKeys,
timeout,
+ waitTimeout,
this,
isInvalidate(),
read,
@@ -767,20 +771,33 @@ public abstract class GridDhtTxLocalAdapter extends
IgniteTxLocalAdapter {
return new GridEmbeddedFuture<>(
fut,
- new PLC1<GridCacheReturn>(ret) {
+ new PLC1<GridCacheReturn>(ret, true,
!CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout)) {
@Override protected GridCacheReturn postLock(GridCacheReturn
ret) throws IgniteCheckedException {
- if (log.isDebugEnabled())
- log.debug("Acquired transaction lock on keys: " +
passedKeys);
-
- postLockWrite(cacheCtx,
- passedKeys,
- ret,
- /*remove*/false,
- /*retval*/false,
- /*read*/read,
- accessTtl,
- CU.empty0(),
- /*computeInvoke*/false);
+ assert fut.error() == null : "Lock future completed with
an error: " + fut.error();
+
+ boolean success = Boolean.TRUE.equals(fut.get());
+
+ ret.success(success);
+
+ if (log.isDebugEnabled()) {
+ if (ret.success())
+ log.debug("Successfully acquired transaction lock
on keys: " + passedKeys);
+ else
+ log.debug("Failed to acquire transaction lock on
keys: " + passedKeys);
+ }
+
+ if (ret.success()) {
+ postLockWrite(cacheCtx,
+ passedKeys,
+ ret,
+ /*remove*/false,
+ /*retval*/false,
+ /*read*/read,
+ accessTtl,
+ CU.empty0(),
+ /*computeInvoke*/false,
+
/*skipIfLockLost*/CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout));
+ }
return ret;
}
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 a214cba09be..71b36a3cef0 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
@@ -816,6 +816,7 @@ public class GridDhtAtomicCache<K, V> extends
GridDhtCacheAdapter<K, V> {
/** {@inheritDoc} */
@Override protected IgniteInternalFuture<Boolean>
lockAllAsync(Collection<KeyCacheObject> keys,
long timeout,
+ long waitTimeout,
@Nullable IgniteTxLocalEx tx,
boolean isInvalidate,
boolean isRead,
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 68b89011bbc..6c4217a2119 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
@@ -639,6 +639,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
@Override public IgniteInternalFuture<Boolean> lockAllAsync(
Collection<KeyCacheObject> keys,
long timeout,
+ long waitTimeout,
@Nullable IgniteTxLocalEx tx,
boolean isInvalidate,
boolean isRead,
@@ -659,6 +660,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
isRead,
retval,
timeout,
+ waitTimeout,
createTtl,
accessTtl,
opCtx != null && opCtx.skipStore(),
@@ -902,6 +904,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
* @param txRead Tx read.
* @param retval Return value flag.
* @param timeout Lock timeout.
+ * @param waitTimeout Lock wait timeout.
* @param createTtl TTL for create operation.
* @param accessTtl TTL for read operation.
* @param skipStore Skip store flag.
@@ -919,6 +922,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
final boolean txRead,
final boolean retval,
final long timeout,
+ final long waitTimeout,
final long createTtl,
final long accessTtl,
final boolean skipStore,
@@ -945,6 +949,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
txRead,
retval,
timeout,
+ waitTimeout,
createTtl,
accessTtl,
skipStore,
@@ -968,6 +973,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
txRead,
retval,
timeout,
+ waitTimeout,
createTtl,
accessTtl,
skipStore,
@@ -990,6 +996,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
* @param txRead Tx read.
* @param retval Return value flag.
* @param timeout Lock timeout.
+ * @param waitTimeout Lock wait timeout.
* @param createTtl TTL for create operation.
* @param accessTtl TTL for read operation.
* @param skipStore Skip store flag.
@@ -1007,6 +1014,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
final boolean txRead,
boolean retval,
final long timeout,
+ long waitTimeout,
final long createTtl,
final long accessTtl,
boolean skipStore,
@@ -1024,6 +1032,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
txRead,
retval,
timeout,
+ waitTimeout,
tx,
threadId,
createTtl,
@@ -1077,7 +1086,7 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
@Override public Exception apply(Boolean b, Exception e) {
if (e != null)
e = U.unwrap(e);
- else if (!b)
+ else if (!b &&
!CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout))
e = new GridCacheLockTimeoutException(ver);
return e;
@@ -1101,7 +1110,8 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
skipStore,
skipReadThrough,
keepBinaryInInterceptor,
- keepBinary);
+ keepBinary,
+ waitTimeout);
return new GridDhtEmbeddedFuture<>(
new C2<GridCacheReturn, Exception, Exception>() {
@@ -1109,8 +1119,8 @@ public class GridDhtColocatedCache<K, V> extends
GridDhtTransactionalCacheAdapte
Exception e) {
if (e != null)
e = U.unwrap(e);
-
- assert !tx.empty();
+ else if (ret != null && !ret.success())
+ e = new GridCacheLockTimeoutException(ver);
return e;
}
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 1e322d8cb56..e27d53a0b85 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
@@ -142,9 +142,12 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
@GridToStringExclude
private volatile LockTimeoutObject timeoutObj;
- /** Lock timeout. */
+ /** Transaction timeout. */
private final long timeout;
+ /** Lock wait timeout. */
+ private final long waitTimeout;
+
/** Transaction. */
@GridToStringExclude
private final GridNearTxLocal tx;
@@ -197,7 +200,8 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
* @param tx Transaction.
* @param read Read flag.
* @param retval Flag to return value or not.
- * @param timeout Lock acquisition timeout.
+ * @param timeout Transaction timeout.
+ * @param waitTimeout Lock wait timeout.
* @param createTtl TTL for create operation.
* @param accessTtl TTL for read operation.
* @param skipStore Skip store flag.
@@ -211,6 +215,7 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
boolean read,
boolean retval,
long timeout,
+ long waitTimeout,
long createTtl,
long accessTtl,
boolean skipStore,
@@ -229,6 +234,7 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
this.read = read;
this.retval = retval;
this.timeout = timeout;
+ this.waitTimeout = waitTimeout;
this.createTtl = createTtl;
this.accessTtl = accessTtl;
this.skipStore = skipStore;
@@ -626,6 +632,9 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
if (err != null)
success = false;
+ if (!success && err == null &&
CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout))
+ return onComplete(false, true, false);
+
return onComplete(success, true);
}
}
@@ -638,6 +647,18 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
* @return {@code True} if complete by this operation.
*/
private boolean onComplete(boolean success, boolean distribute) {
+ return onComplete(success, distribute, !success);
+ }
+
+ /**
+ * Completeness callback.
+ *
+ * @param success {@code True} if lock was acquired.
+ * @param distribute {@code True} if need to distribute lock removal in
case of failure.
+ * @param rollback {@code True} if should rollback tx on failure.
+ * @return {@code True} if complete by this operation.
+ */
+ private boolean onComplete(boolean success, boolean distribute, boolean
rollback) {
if (log.isDebugEnabled()) {
log.debug("Received onComplete(..) callback [success=" + success +
", distribute=" + distribute +
", fut=" + this + ']');
@@ -646,13 +667,13 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
if (!DONE_UPD.compareAndSet(this, 0, 1))
return false;
- if (!success)
+ if (!success && rollback)
undoLocks(distribute, true);
if (tx != null) {
cctx.tm().txContext(tx);
- if (success)
+ if (!rollback)
tx.clearLockFuture(this);
}
@@ -768,7 +789,7 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
if (isDone()) // Possible due to async rollback.
return;
- if (timeout > 0) {
+ if (lockTimeout() > 0) {
timeoutObj = new LockTimeoutObject();
cctx.time().addTimeoutObject(timeoutObj);
@@ -1083,6 +1104,7 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
isolation(),
isInvalidate(),
timeout,
+ waitTimeout,
mappedKeys.size(),
inTx() ? tx.size() : mappedKeys.size(),
inTx() && tx.syncMode() == FULL_SYNC,
@@ -1258,6 +1280,7 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
read,
retval,
timeout,
+ waitTimeout,
createTtl,
accessTtl,
skipStore,
@@ -1477,6 +1500,13 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
return false;
}
+ /**
+ * @return Timeout value for this lock future.
+ */
+ private long lockTimeout() {
+ return CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout) ?
waitTimeout : timeout;
+ }
+
/**
* Lock request timeout object.
*/
@@ -1485,7 +1515,7 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
* Default constructor.
*/
LockTimeoutObject() {
- super(timeout);
+ super(lockTimeout());
}
/** Requested keys. */
@@ -1496,6 +1526,20 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
if (log.isDebugEnabled())
log.debug("Timed out waiting for lock response: " + this);
+ if (CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout)) {
+ synchronized (GridDhtColocatedLockFuture.this) {
+ requestedKeys = requestedKeys0();
+
+ clear(); // Stop response processing.
+ }
+
+ synchronized (this) {
+ onComplete(false, true, false);
+ }
+
+ return;
+ }
+
if (inTx()) {
if (cctx.tm().deadlockDetectionEnabled()) {
synchronized (GridDhtColocatedLockFuture.this) {
@@ -1660,6 +1704,12 @@ public final class GridDhtColocatedLockFuture extends
GridCacheCompoundIdentityF
return;
}
+ if (!res.lockAcquired()) {
+ onDone(false);
+
+ return;
+ }
+
if (res.clientRemapVersion() != null) {
assert cctx.kernalContext().clientNode();
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
index 0a9b5a7c658..366f937e380 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearAtomicCache.java
@@ -594,6 +594,7 @@ public class GridNearAtomicCache<K, V> extends
GridNearCacheAdapter<K, V> {
/** {@inheritDoc} */
@Override protected IgniteInternalFuture<Boolean>
lockAllAsync(Collection<KeyCacheObject> keys,
long timeout,
+ long waitTimeout,
@Nullable IgniteTxLocalEx tx,
boolean isInvalidate,
boolean isRead,
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 a5db4ea3e61..20923899b32 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
@@ -124,13 +124,19 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
/** Timed out flag. */
private volatile boolean timedOut;
+ /** Transaction lock timeout flag. */
+ private volatile boolean txLockTimedOut;
+
/** Timeout object. */
@GridToStringExclude
private volatile LockTimeoutObject timeoutObj;
- /** Lock timeout. */
+ /** Transaction timeout. */
private final long timeout;
+ /** Lock wait timeout. */
+ private final long waitTimeout;
+
/** Transaction. */
@GridToStringExclude
private final GridNearTxLocal tx;
@@ -182,7 +188,8 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
* @param tx Transaction.
* @param read Read flag.
* @param retval Flag to return value or not.
- * @param timeout Lock acquisition timeout.
+ * @param timeout Transaction timeout.
+ * @param waitTimeout Lock wait timeout.
* @param createTtl TTL for create operation.
* @param accessTtl TTL for read operation.
* @param skipStore skipStore
@@ -198,6 +205,7 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
boolean read,
boolean retval,
long timeout,
+ long waitTimeout,
long createTtl,
long accessTtl,
boolean skipStore,
@@ -217,6 +225,7 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
this.read = read;
this.retval = retval;
this.timeout = timeout;
+ this.waitTimeout = waitTimeout;
this.createTtl = createTtl;
this.accessTtl = accessTtl;
this.skipStore = skipStore;
@@ -347,7 +356,7 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
threadId,
lockVer,
topVer,
- timeout,
+ lockTimeout(),
!inTx(),
inTx(),
implicitSingleTx(),
@@ -362,11 +371,16 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
entries.add(entry);
- if (c == null && timeout < 0) {
+ if (c == null && lockTimeout() < 0) {
if (log.isDebugEnabled())
log.debug("Failed to acquire lock with negative timeout: " +
entry);
- onFailed(false);
+ if (CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout)) {
+ onComplete(false, true, false);
+ }
+ else {
+ onFailed(false);
+ }
return null;
}
@@ -696,7 +710,7 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
log.debug("Received onDone(..) callback [success=" + success + ",
err=" + err + ", fut=" + this + ']');
if (inTx() && cctx.tm().deadlockDetectionEnabled() &&
- (this.err instanceof IgniteTxTimeoutCheckedException || timedOut))
+ (this.err instanceof IgniteTxTimeoutCheckedException ||
txLockTimedOut))
return false;
// If locks were not acquired yet, delay completion.
@@ -709,6 +723,9 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
if (err != null)
success = false;
+ if (!success && err == null &&
CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout))
+ return onComplete(false, true, false);
+
return onComplete(success, true);
}
@@ -720,6 +737,18 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
* @return {@code True} if complete by this operation.
*/
private boolean onComplete(boolean success, boolean distribute) {
+ return onComplete(success, distribute, !success);
+ }
+
+ /**
+ * Completeness callback.
+ *
+ * @param success {@code True} if lock was acquired.
+ * @param distribute {@code True} if need to distribute lock removal in
case of failure.
+ * @param rollback {@code True} if should rollback tx on failure.
+ * @return {@code True} if complete by this operation.
+ */
+ private boolean onComplete(boolean success, boolean distribute, boolean
rollback) {
if (log.isDebugEnabled()) {
log.debug("Received onComplete(..) callback [success=" + success +
", distribute=" + distribute +
", fut=" + this + ']');
@@ -728,13 +757,13 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
if (!DONE_UPD.compareAndSet(this, 0, 1))
return false;
- if (!success)
+ if (!success && rollback)
undoLocks(distribute, true);
if (tx != null) {
cctx.tm().txContext(tx);
- if (success)
+ if (!rollback)
tx.clearLockFuture(this);
}
@@ -798,7 +827,7 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
if (isDone()) // Possible due to async rollback.
return;
- if (timeout > 0) {
+ if (lockTimeout() > 0) {
timeoutObj = new LockTimeoutObject();
cctx.time().addTimeoutObject(timeoutObj);
@@ -1069,6 +1098,7 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
isolation(),
isInvalidate(),
timeout,
+ waitTimeout,
mappedKeys.size(),
inTx() ? tx.size() :
mappedKeys.size(),
inTx() && tx.syncMode() ==
FULL_SYNC,
@@ -1220,6 +1250,9 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
return false;
}
+ if (!res.lockAcquired())
+ return false;
+
if (log.isDebugEnabled())
log.debug("Acquired lock for local DHT mapping
[locId=" + cctx.nodeId() +
", mappedKeys=" + mappedKeys + ", fut=" +
GridNearLockFuture.this + ']');
@@ -1397,6 +1430,13 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
return topEx;
}
+ /**
+ * @return Timeout value for this lock future.
+ */
+ private long lockTimeout() {
+ return CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout) ?
waitTimeout : timeout;
+ }
+
/**
* Lock request timeout object.
*/
@@ -1405,7 +1445,7 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
* Default constructor.
*/
LockTimeoutObject() {
- super(timeout);
+ super(lockTimeout());
}
/** Requested keys. */
@@ -1418,6 +1458,22 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
timedOut = true;
+ if (CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout)) {
+ synchronized (GridNearLockFuture.this) {
+ requestedKeys = requestedKeys0();
+
+ clear(); // Stop response processing.
+ }
+
+ synchronized (this) {
+ onComplete(false, true, false);
+ }
+
+ return;
+ }
+
+ txLockTimedOut = true;
+
if (inTx()) {
if (cctx.tm().deadlockDetectionEnabled()) {
synchronized (GridNearLockFuture.this) {
@@ -1579,6 +1635,12 @@ public final class GridNearLockFuture extends
GridCacheCompoundIdentityFuture<Bo
return;
}
+ if (!res.lockAcquired()) {
+ onDone(false);
+
+ return;
+ }
+
if (res.clientRemapVersion() != null) {
assert cctx.kernalContext().clientNode();
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 c976f22df27..46fae249ebd 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
@@ -79,6 +79,10 @@ public class GridNearLockRequest extends
GridDistributedLockRequest {
@Order(7)
String txLbl;
+ /** Lock wait timeout. */
+ @Order(8)
+ long waitTimeout;
+
/**
* Empty constructor.
*/
@@ -98,7 +102,8 @@ public class GridNearLockRequest extends
GridDistributedLockRequest {
* @param retVal Return value flag.
* @param isolation Transaction isolation.
* @param isInvalidate Invalidation flag.
- * @param timeout Lock timeout.
+ * @param timeout Transaction timeout.
+ * @param waitTimeout Lock wait timeout.
* @param keyCnt Number of keys.
* @param txSize Expected transaction size.
* @param syncCommit Synchronous commit flag.
@@ -123,6 +128,7 @@ public class GridNearLockRequest extends
GridDistributedLockRequest {
TransactionIsolation isolation,
boolean isInvalidate,
long timeout,
+ long waitTimeout,
int keyCnt,
int txSize,
boolean syncCommit,
@@ -162,6 +168,7 @@ public class GridNearLockRequest extends
GridDistributedLockRequest {
this.taskNameHash = taskNameHash;
this.createTtl = createTtl;
this.accessTtl = accessTtl;
+ this.waitTimeout = waitTimeout;
this.txLbl = txLbl;
@@ -207,6 +214,13 @@ public class GridNearLockRequest extends
GridDistributedLockRequest {
return isFlag(FIRST_CLIENT_REQ_FLAG_MASK);
}
+ /**
+ * @return Lock wait timeout.
+ */
+ public long waitTimeout() {
+ return waitTimeout;
+ }
+
/**
* @return Topology version.
*/
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
index 043e4779c49..b179d094d8c 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockResponse.java
@@ -68,6 +68,10 @@ public class GridNearLockResponse extends
GridDistributedLockResponse {
@Order(6)
boolean compatibleRemapVer;
+ /** {@code True} if requested locks were acquired. */
+ @Order(7)
+ boolean lockAcquired = true;
+
/**
* Empty constructor.
*/
@@ -124,6 +128,20 @@ public class GridNearLockResponse extends
GridDistributedLockResponse {
return compatibleRemapVer;
}
+ /**
+ * @return {@code True} if requested locks were acquired.
+ */
+ public boolean lockAcquired() {
+ return lockAcquired;
+ }
+
+ /**
+ * @param lockAcquired {@code True} if requested locks were acquired.
+ */
+ public void lockAcquired(boolean lockAcquired) {
+ this.lockAcquired = lockAcquired;
+ }
+
/**
* @return Pending versions that are less than {@link #version()}.
*/
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
index 757cc77b21c..c59b3a557d1 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearPessimisticTxPrepareFuture.java
@@ -378,7 +378,8 @@ public class GridNearPessimisticTxPrepareFuture extends
GridNearTxPrepareFutureA
GridNearTxPrepareRequest req = createRequest(txNodes,
m,
timeout,
- m.reads(),
+ // Read entries do not make sense in the prepare phase for
pessimistic transactions.
+ List.of(),
m.writes());
final MiniFuture fut = new MiniFuture(m, ++miniId);
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 d79cf19d9ce..7074bda1b12 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
@@ -289,6 +289,7 @@ public class GridNearTransactionalCache<K, V> extends
GridNearCacheAdapter<K, V>
@Override protected IgniteInternalFuture<Boolean> lockAllAsync(
Collection<KeyCacheObject> keys,
long timeout,
+ long waitTimeout,
IgniteTxLocalEx tx,
boolean isInvalidate,
boolean isRead,
@@ -305,6 +306,7 @@ public class GridNearTransactionalCache<K, V> extends
GridNearCacheAdapter<K, V>
isRead,
retval,
timeout,
+ waitTimeout,
createTtl,
accessTtl,
opCtx != null && opCtx.skipStore(),
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 dffbf776c51..f0f4ddaaf26 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
@@ -650,6 +650,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter
implements GridTimeou
log.debug("Before acquiring transaction lock for put on
key: " + enlisted);
IgniteInternalFuture<Boolean> fut =
cacheCtx.cache().txLockAsync(enlisted,
+ timeout,
timeout,
this,
/*read*/entryProc != null, // Needed to force load from
store.
@@ -826,6 +827,7 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter
implements GridTimeou
log.debug("Before acquiring transaction lock for put on
keys: " + enlisted);
IgniteInternalFuture<Boolean> fut =
cacheCtx.cache().txLockAsync(enlisted,
+ timeout,
timeout,
this,
/*read*/invokeVals != null, // Needed to force load from
store.
@@ -1736,6 +1738,7 @@ public class GridNearTxLocal extends
GridDhtTxLocalAdapter implements GridTimeou
log.debug("Before acquiring transaction lock for remove on
keys: " + enlisted);
IgniteInternalFuture<Boolean> fut =
cacheCtx.cache().txLockAsync(enlisted,
+ timeout,
timeout,
this,
false,
@@ -1930,6 +1933,7 @@ public class GridNearTxLocal extends
GridDhtTxLocalAdapter implements GridTimeou
return new GridFinishedFuture<>(timeoutException());
IgniteInternalFuture<Boolean> fut =
cacheCtx.cache().txLockAsync(lockKeys,
+ timeout,
timeout,
this,
true,
@@ -3233,6 +3237,23 @@ public class GridNearTxLocal extends
GridDhtTxLocalAdapter implements GridTimeou
removeEntryFromMappings(entry, nearMap);
}
+ /**
+ * Removes transaction entries and releases their acquired transactional
locks.
+ *
+ * @param entries Entries to remove and unlock.
+ */
+ public void removeAndUnlockTxEntries(Collection<IgniteTxEntry> entries) {
+ if (F.isEmpty(entries))
+ return;
+
+ for (IgniteTxEntry entry : entries) {
+ txState().removeEntry(entry.txKey());
+ removeEntryMappings(entry);
+ }
+
+ unlockTxEntries(entries);
+ }
+
/**
* @param entry Entry.
*/
@@ -3294,7 +3315,16 @@ public class GridNearTxLocal extends
GridDhtTxLocalAdapter implements GridTimeou
else if (cacheCtx.cache().isColocated()) {
UUID nodeId = entry.nodeId();
- if (nodeId == null || cctx.localNodeId().equals(nodeId))
+ if (nodeId == null) {
+ ClusterNode primary =
cacheCtx.affinity().primaryByKey(entry.key(), topologyVersion());
+
+ if (primary == null)
+ continue;
+
+ nodeId = primary.id();
+ }
+
+ if (cctx.localNodeId().equals(nodeId))
colocatedLocKeys.computeIfAbsent(cacheCtx, k -> new
ArrayList<>()).add(entry.key());
else {
colocatedRmtKeys
@@ -4223,6 +4253,7 @@ public class GridNearTxLocal extends
GridDhtTxLocalAdapter implements GridTimeou
* @param skipReadThrough Skip read-through cache store flag.
* @param keepBinaryInInterceptor Handle binary in interceptor operation
flag.
* @param keepBinary Keep binary flag.
+ * @param waitTimeout Lock wait timeout.
* @return Future with respond.
*/
public <K> IgniteInternalFuture<GridCacheReturn>
lockAllAsync(GridCacheContext cacheCtx,
@@ -4234,7 +4265,8 @@ public class GridNearTxLocal extends
GridDhtTxLocalAdapter implements GridTimeou
boolean skipStore,
boolean skipReadThrough,
boolean keepBinaryInInterceptor,
- boolean keepBinary) {
+ boolean keepBinary,
+ long waitTimeout) {
assert pessimistic();
try {
@@ -4261,6 +4293,7 @@ public class GridNearTxLocal extends
GridDhtTxLocalAdapter implements GridTimeou
IgniteInternalFuture<Boolean> fut =
cacheCtx.colocated().lockAllAsyncInternal(keys,
timeout,
+ waitTimeout,
this,
isInvalidate(),
read,
@@ -4275,10 +4308,20 @@ public class GridNearTxLocal extends
GridDhtTxLocalAdapter implements GridTimeou
return new GridEmbeddedFuture<>(
fut,
- new PLC1<GridCacheReturn>(ret, false) {
- @Override protected GridCacheReturn postLock(GridCacheReturn
ret) {
- if (log.isDebugEnabled())
- log.debug("Acquired transaction lock on keys: " +
keys);
+ new PLC1<GridCacheReturn>(ret, false,
!CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout)) {
+ @Override protected GridCacheReturn postLock(GridCacheReturn
ret) throws IgniteCheckedException {
+ assert fut.error() == null : "Lock future completed with
an error: " + fut.error();
+
+ boolean success = Boolean.TRUE.equals(fut.get());
+
+ ret.success(success);
+
+ if (log.isDebugEnabled()) {
+ if (ret.success())
+ log.debug("Successfully acquired transaction lock
on keys: " + keys);
+ else
+ log.debug("Failed to acquire transaction lock on
keys: " + keys);
+ }
return ret;
}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
index cbb73652e03..bbc6e282fd5 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxLocalAdapter.java
@@ -1089,6 +1089,36 @@ public abstract class IgniteTxLocalAdapter extends
IgniteTxAdapter implements Ig
long accessTtl,
CacheEntryPredicate[] filter,
boolean computeInvoke
+ ) throws IgniteCheckedException {
+ postLockWrite(cacheCtx, keys, ret, rmv, retval, read, accessTtl,
filter, computeInvoke, false);
+ }
+
+ /**
+ * Post lock processing for put or remove.
+ *
+ * @param cacheCtx Context.
+ * @param keys Keys.
+ * @param ret Return value.
+ * @param rmv {@code True} if remove.
+ * @param retval Flag to return value or not.
+ * @param read {@code True} if read.
+ * @param accessTtl TTL for read operation.
+ * @param filter Filter to check entries.
+ * @param computeInvoke If {@code true} computes return value for invoke
operation.
+ * @param skipIfLockLost Return unsuccessful result if a separate lock
wait timeout has removed the lock.
+ * @throws IgniteCheckedException If error.
+ */
+ protected final void postLockWrite(
+ GridCacheContext cacheCtx,
+ Iterable<KeyCacheObject> keys,
+ GridCacheReturn ret,
+ boolean rmv,
+ boolean retval,
+ boolean read,
+ long accessTtl,
+ CacheEntryPredicate[] filter,
+ boolean computeInvoke,
+ boolean skipIfLockLost
) throws IgniteCheckedException {
for (KeyCacheObject k : keys) {
IgniteTxEntry txEntry = entry(cacheCtx.txKey(k));
@@ -1101,7 +1131,15 @@ public abstract class IgniteTxLocalAdapter extends
IgniteTxAdapter implements Ig
GridCacheEntryEx cached = txEntry.cached();
try {
- assert cached.detached() ||
cached.lockedLocally(xidVersion()) || isRollbackOnly() :
+ boolean ownsLock = cached.detached() ||
cached.lockedLocally(xidVersion());
+
+ if (!ownsLock && skipIfLockLost) {
+ ret.success(false);
+
+ return;
+ }
+
+ assert ownsLock || isRollbackOnly() :
"Transaction lock is not acquired [entry=" + cached +
", tx=" + this +
", nodeId=" + cctx.localNodeId() + ", threadId=" +
threadId + ']';
@@ -1602,9 +1640,10 @@ public abstract class IgniteTxLocalAdapter extends
IgniteTxAdapter implements Ig
/**
* @param arg Argument.
* @param commit Commit flag.
+ * @param rollback Rollback flag.
*/
- protected PLC1(T arg, boolean commit) {
- super(arg, commit);
+ protected PLC1(T arg, boolean commit, boolean rollback) {
+ super(arg, commit, rollback);
}
}
@@ -1635,13 +1674,16 @@ public abstract class IgniteTxLocalAdapter extends
IgniteTxAdapter implements Ig
/** Commit flag. */
private final boolean commit;
+ /** Rollback when a lock is not acquired. */
+ private final boolean rollback;
+
/**
* Creates a Post-Lock closure that will pass the argument given to
the {@code postLock} method.
*
* @param arg Argument for {@code postLock}.
*/
protected PostLockClosure1(T arg) {
- this(arg, true);
+ this(arg, true, true);
}
/**
@@ -1649,10 +1691,12 @@ public abstract class IgniteTxLocalAdapter extends
IgniteTxAdapter implements Ig
*
* @param arg Argument for {@code postLock}.
* @param commit Flag indicating whether commit should be done after
postLock.
+ * @param rollback Flag indicating whether rollback should be done if
lock is not acquired.
*/
- protected PostLockClosure1(T arg, boolean commit) {
+ protected PostLockClosure1(T arg, boolean commit, boolean rollback) {
this.arg = arg;
this.commit = commit;
+ this.rollback = rollback;
}
/** {@inheritDoc} */
@@ -1670,7 +1714,7 @@ public abstract class IgniteTxLocalAdapter extends
IgniteTxAdapter implements Ig
throw new GridClosureException(e);
}
- if (deadlockErr != null || !locked) {
+ if (deadlockErr != null || (!locked && rollback)) {
setRollbackOnly();
final GridClosureException ex = new GridClosureException(
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
index 7f9298c352e..e4a56ac880e 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxManager.java
@@ -1508,10 +1508,7 @@ public class IgniteTxManager extends
GridCacheSharedManagerAdapter {
* @return {@code True} if transaction read entries should be unlocked.
*/
private boolean unlockReadEntries(IgniteInternalTx tx) {
- if (tx.pessimistic())
- return !tx.readCommitted();
- else
- return tx.serializable();
+ return tx.pessimistic() || tx.serializable();
}
/**
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheOperationContextTransactionalLockTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheOperationContextTransactionalLockTest.java
new file mode 100644
index 00000000000..76e36e6ab54
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheOperationContextTransactionalLockTest.java
@@ -0,0 +1,141 @@
+/*
+ * 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 org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntry;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxEntry;
+import org.apache.ignite.internal.processors.cache.transactions.IgniteTxKey;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.junit.Test;
+
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+
+/** Tests operation context propagation when a versioned cache entry is locked
in a transaction. */
+public class CacheOperationContextTransactionalLockTest extends
GridCommonAbstractTest {
+ /** */
+ private static final int KEY = 1;
+
+ /** */
+ private static Ignite ignite;
+
+ /** */
+ private static IgniteCache<Integer, Integer> cache;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ ignite = startGrid(0);
+
+ cache = ignite.createCache(new CacheConfiguration<Integer,
Integer>(DEFAULT_CACHE_NAME)
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL));
+
+ cache.put(KEY, KEY);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ ignite = null;
+ cache = null;
+
+ super.afterTestsStopped();
+ }
+
+ /** Checks the defaults used when there is no operation context. */
+ @Test
+ public void testDefaultOperationContext() throws Exception {
+ assertTxEntryFlags(internalCache(), false, false, false, false);
+ }
+
+ /** Checks propagation of the skip-store flag. */
+ @Test
+ public void testSkipStoreOperationContext() throws Exception {
+ assertTxEntryFlags(internalCache().withSkipStore(), true, false,
false, false);
+ }
+
+ /** Checks propagation of the skip-read-through flag. */
+ @Test
+ public void testSkipReadThroughOperationContext() throws Exception {
+ assertTxEntryFlags(internalCache().withSkipReadThrough(), false, true,
false, false);
+ }
+
+ /** Checks propagation of the keep-binary-in-interceptor flag. */
+ @Test
+ public void testKeepBinaryInInterceptorOperationContext() throws Exception
{
+ assertTxEntryFlags(internalCache().withKeepBinaryInInterceptor(),
false, false, true, false);
+ }
+
+ /** Checks propagation of the keep-binary flag. */
+ @Test
+ public void testKeepBinaryOperationContext() throws Exception {
+ assertTxEntryFlags(internalCache().keepBinary(), false, false, true,
true);
+ }
+
+ /**
+ * Locks a versioned entry and checks flags stored in its transaction
entry.
+ *
+ * @param internalCache Internal cache projection used to acquire the lock.
+ * @param skipStore Expected skip-store flag.
+ * @param skipReadThrough Expected skip-read-through flag.
+ * @param keepBinaryInInterceptor Expected keep-binary-in-interceptor flag.
+ * @param keepBinary Expected keep-binary flag.
+ * @throws Exception If failed.
+ */
+ private void assertTxEntryFlags(
+ IgniteInternalCache<Integer, Integer> internalCache,
+ boolean skipStore,
+ boolean skipReadThrough,
+ boolean keepBinaryInInterceptor,
+ boolean keepBinary
+ ) throws Exception {
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+
+ assertNotNull(entry);
+ assertNotNull(entry.version());
+
+ try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertTrue(internalCache.lockTxEntry(entry, 0));
+
+ GridCacheContext<Integer, Integer> ctx = internalCache.context();
+ IgniteTxKey txKey = ctx.txKey(ctx.toCacheKeyObject(KEY));
+ IgniteTxEntry txEntry = internalCache.tx().entry(txKey);
+
+ assertNotNull(txEntry);
+ assertEquals(skipStore, txEntry.skipStore());
+ assertEquals(skipReadThrough, txEntry.skipReadThrough());
+ assertEquals(keepBinaryInInterceptor,
txEntry.keepBinaryInInterceptor());
+ assertEquals(keepBinary, txEntry.keepBinary());
+
+ tx.commit();
+ }
+ }
+
+ /** Returns the cache's internal proxy. */
+ @SuppressWarnings("unchecked")
+ private static IgniteInternalCache<Integer, Integer> internalCache() {
+ return cache.unwrap(IgniteCacheProxy.class).internalProxy();
+ }
+}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheVersionedEntryTransactionalLockTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheVersionedEntryTransactionalLockTest.java
new file mode 100644
index 00000000000..41acd7a472c
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheVersionedEntryTransactionalLockTest.java
@@ -0,0 +1,514 @@
+/*
+ * 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.Collection;
+import java.util.List;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntry;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionIsolation;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static
org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/**
+ * Tests transactional locks acquired only for unchanged cache entry versions.
+ */
+@RunWith(Parameterized.class)
+public class CacheVersionedEntryTransactionalLockTest extends
GridCommonAbstractTest {
+ /** */
+ private static Ignite ignite0;
+
+ /** */
+ private static Ignite ignite1;
+
+ /** */
+ private static Ignite client;
+
+ /** */
+ @Parameterized.Parameter(0)
+ public boolean useNearCache;
+
+ /** */
+ @Parameterized.Parameter(1)
+ public int backups;
+
+ /** */
+ @Parameterized.Parameter(2)
+ public boolean replicated;
+
+ /** */
+ @Parameterized.Parameter(3)
+ public boolean batch;
+
+ /**
+ * Returns data for test.
+ *
+ * @return Test parameters.
+ */
+ @Parameterized.Parameters(name = "useNearCache={0}, backups={1},
replicated={2}, batch={3}")
+ public static Collection<Object[]> testData() {
+ return List.of(new Object[][] {
+ {false, 0, false, false},
+ {false, 0, false, true},
+ {false, 0, true, false},
+ {false, 0, true, true},
+ {false, 1, false, false},
+ {false, 1, false, true},
+ {false, 1, true, false},
+ {false, 1, true, true},
+ {false, 2, false, false},
+ {false, 2, false, true},
+ {false, 2, true, false},
+ {false, 2, true, true},
+ {true, 0, false, false},
+ {true, 0, false, true},
+ {true, 0, true, false},
+ {true, 0, true, true},
+ {true, 1, false, false},
+ {true, 1, false, true},
+ {true, 1, true, false},
+ {true, 1, true, true},
+ {true, 2, false, false},
+ {true, 2, false, true},
+ {true, 2, true, false},
+ {true, 2, true, true}
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ ignite0 = startGridsMultiThreaded(4);
+
+ awaitPartitionMapExchange();
+
+ ignite1 = grid(1);
+ client = startClientGrid();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ ignite0 = null;
+ ignite1 = null;
+ client = null;
+
+ super.afterTestsStopped();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ ignite0.destroyCache(DEFAULT_CACHE_NAME);
+
+ super.afterTest();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String
igniteInstanceName) throws Exception {
+ return super.getConfiguration(igniteInstanceName)
+ .setConsistentId(igniteInstanceName);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return 120_000;
+ }
+
+ /**
+ * Creates transactional cache.
+ *
+ * @param ignite Node.
+ * @return Transactional cache.
+ */
+ private IgniteCache<Integer, Integer> transactionalCache(Ignite ignite) {
+ CacheConfiguration<?, ?> ccfg =
+ new CacheConfiguration<>(DEFAULT_CACHE_NAME)
+
.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC)
+ .setNearConfiguration(useNearCache ? new
NearCacheConfiguration<>() : null)
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ .setCacheMode(replicated ? CacheMode.REPLICATED :
CacheMode.PARTITIONED)
+ .setBackups(backups);
+
+ return (IgniteCache<Integer, Integer>)ignite.createCache(ccfg);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockBeforePutFromClientTest() throws Exception {
+ IgniteCache<Integer, Integer> cache = transactionalCache(client);
+
+ int key = 42;
+
+ checkLockBeforePut(cache, key, READ_COMMITTED);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockBeforePutLocalKeyTest() throws Exception {
+ IgniteCache<Integer, Integer> cache = transactionalCache(ignite0);
+
+ int key = primaryKey(cache);
+
+ checkLockBeforePut(cache, key, READ_COMMITTED);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockBeforePutRemoteKeyTest() throws Exception {
+ IgniteCache<Integer, Integer> cache = transactionalCache(ignite0);
+
+ int key = primaryKey(ignite1.cache(DEFAULT_CACHE_NAME));
+
+ checkLockBeforePut(cache, key, REPEATABLE_READ);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockBeforePutLocalKeyRepeatableReadTest() throws Exception
{
+ IgniteCache<Integer, Integer> cache = transactionalCache(ignite0);
+
+ int key = primaryKey(cache);
+
+ checkLockBeforePut(cache, key, REPEATABLE_READ);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockBeforePutRemoteKeyRepeatableReadTest() throws
Exception {
+ IgniteCache<Integer, Integer> cache = transactionalCache(ignite0);
+
+ int key = primaryKey(ignite1.cache(DEFAULT_CACHE_NAME));
+
+ checkLockBeforePut(cache, key, REPEATABLE_READ);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testEntryVersionDoesNotChangeWhenEntryIsNotUpdated() throws
Exception {
+ IgniteCache<Integer, Integer> cache = transactionalCache(ignite0);
+
+ int locKey = primaryKey(cache);
+ int remoteKey = primaryKey(ignite1.cache(DEFAULT_CACHE_NAME));
+
+ cache.put(locKey, 0);
+ cache.put(remoteKey, 0);
+
+ CacheEntry<Integer, Integer> locEntry = cache.getEntry(locKey);
+ CacheEntry<Integer, Integer> remoteEntry = cache.getEntry(remoteKey);
+
+ try (Transaction tx = ignite0.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ if (batch) {
+ assertTrue(acquireLockForEntries(cache, List.of(locEntry,
remoteEntry), 0));
+ }
+ else {
+ assertTrue(acquireLockForEntry(cache, locEntry, 0));
+ assertTrue(acquireLockForEntry(cache, remoteEntry, 0));
+ }
+
+ tx.commit();
+ }
+
+ assertEquals(locEntry.version(), cache.getEntry(locKey).version());
+ assertEquals(remoteEntry.version(),
cache.getEntry(remoteKey).version());
+
+ try (Transaction tx = ignite0.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ if (batch) {
+ assertTrue(acquireLockForEntries(cache, List.of(locEntry,
remoteEntry), 0));
+ }
+ else {
+ assertTrue(acquireLockForEntry(cache, locEntry, 0));
+ assertTrue(acquireLockForEntry(cache, remoteEntry, 0));
+ }
+
+ tx.rollback();
+ }
+
+ assertEquals(locEntry.version(), cache.getEntry(locKey).version());
+ assertEquals(remoteEntry.version(),
cache.getEntry(remoteKey).version());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void
testVersionedEntryLockReturnsFalseWhenEntryIsLockedByAnotherTransactionNoWait()
throws Exception {
+ checkReturningWhenCanNotWaitForLock(-1, false, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void
testVersionedEntryLockReturnsFalseWhenEntryIsLockedByAnotherTransaction()
throws Exception {
+ checkReturningWhenCanNotWaitForLock(200, false, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void
testVersionedEntryLockReturnsFalseWhenLocalEntryIsLockedByAnotherTransaction()
throws Exception {
+ checkReturningWhenCanNotWaitForLock(200, false, true);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void
testVersionedEntryLockReturnsFalseWhenEntryIsLockedByAnotherTransactionAndCommit()
throws Exception {
+ checkReturningWhenCanNotWaitForLock(200, true, false);
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void
testVersionedEntryLockReturnsFalseWhenLocalEntryIsLockedByAnotherTransactionAndCommit()
throws Exception {
+ checkReturningWhenCanNotWaitForLock(200, true, true);
+ }
+
+ /**
+ * Checks that lock acquisition can be retried in the same transaction
after the competing transaction finishes.
+ *
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testVersionedEntryLockCanBeRetriedAfterWaitTimeout() throws
Exception {
+ Ignite holder = ignite0;
+ Ignite initiator = ignite1;
+
+ IgniteCache<Integer, Integer> holderCache = transactionalCache(holder);
+ IgniteCache<Integer, Integer> cache =
initiator.cache(DEFAULT_CACHE_NAME);
+
+ int key = primaryKey(holderCache);
+
+ holderCache.put(key, 0);
+
+ CacheEntry<Integer, Integer> entry = cache.getEntry(key);
+
+ try (Transaction holderTx = holder.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ holderCache.put(key, 42);
+
+ try (Transaction tx =
initiator.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
+ if (batch)
+ assertFalse(acquireLockForEntries(cache, List.of(entry),
200));
+ else
+ assertFalse(acquireLockForEntry(cache, entry, 200));
+
+ holderTx.rollback();
+
+ if (batch)
+ assertTrue(acquireLockForEntries(cache, List.of(entry),
5_000));
+ else
+ assertTrue(acquireLockForEntry(cache, entry, 5_000));
+
+ cache.put(key, 1);
+
+ tx.commit();
+ }
+ }
+
+ assertEquals(1, cache.get(key).intValue());
+ }
+
+ /**
+ * Checks that the lock entry method returns {@code false} when can not
wait for lock.
+ *
+ * @param timeout Timeout.
+ * @param commit Whether to commit the transaction.
+ * @param locForLocal Whether the contended key should be local to the
lock initiator.
+ * @throws IgniteCheckedException If failed.
+ */
+ private void checkReturningWhenCanNotWaitForLock(long timeout, boolean
commit, boolean locForLocal) throws IgniteCheckedException {
+ Ignite holder = ignite0;
+ Ignite initiator = ignite1;
+
+ IgniteCache<Integer, Integer> holderCache = transactionalCache(holder);
+ IgniteCache<Integer, Integer> cache =
initiator.cache(DEFAULT_CACHE_NAME);
+
+ List<Integer> firstKeySet = locForLocal ? primaryKeys(cache, 2) :
primaryKeys(holderCache, 2);
+ Integer concurentLockedKey = firstKeySet.get(0);
+ Integer txKey1 = firstKeySet.get(1);
+ Integer txKey2 = locForLocal ? primaryKey(holderCache) :
primaryKey(cache);
+
+ holderCache.put(concurentLockedKey, 0);
+ holderCache.put(txKey1, 0);
+ holderCache.put(txKey2, 0);
+
+ CacheEntry<Integer, Integer> entry =
cache.getEntry(concurentLockedKey);
+
+ try (Transaction holderTx = holder.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ holderCache.put(concurentLockedKey, 42);
+
+ try (Transaction tx =
initiator.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
+ long startWaiting = U.currentTimeMillis();
+
+ if (batch) {
+ assertFalse(acquireLockForEntries(cache, List.of(
+ cache.getEntry(txKey1),
+ entry,
+ cache.getEntry(txKey2)
+ ), timeout));
+
+ assertTrue(acquireLockForEntries(cache, List.of(
+ cache.getEntry(txKey1),
+ cache.getEntry(txKey2)
+ ), timeout));
+ }
+ else {
+ assertTrue(acquireLockForEntry(cache,
cache.getEntry(txKey1), timeout));
+ assertTrue(acquireLockForEntry(cache,
cache.getEntry(txKey2), timeout));
+
+ assertFalse(acquireLockForEntry(cache, entry, timeout));
+ }
+
+ long waitTime = U.currentTimeMillis() - startWaiting;
+
+ assertTrue("Waited for lock for " + waitTime + " ms but
timeout is " + timeout + " ms.",
+ waitTime >= timeout);
+
+ cache.put(txKey1, 42);
+ cache.put(txKey2, 42);
+
+ if (commit)
+ tx.commit();
+ }
+
+ holderTx.rollback();
+ }
+
+ assertEquals(0, holderCache.get(concurentLockedKey).intValue());
+ assertEquals(0, cache.get(concurentLockedKey).intValue());
+
+ if (commit) {
+ assertEquals(42, holderCache.get(txKey1).intValue());
+ assertEquals(42, cache.get(txKey2).intValue());
+ }
+ else {
+ assertEquals(0, holderCache.get(txKey1).intValue());
+ assertEquals(0, cache.get(txKey2).intValue());
+ }
+ }
+
+ /**
+ * Checks locking an entry before updating it.
+ *
+ * @param cache Cache.
+ * @param key Key.
+ * @param txIsolation Transaction isolation.
+ * @throws IgniteCheckedException If failed.
+ */
+ private void checkLockBeforePut(
+ IgniteCache<Integer, Integer> cache,
+ int key,
+ TransactionIsolation txIsolation
+ ) throws IgniteCheckedException {
+ cache.put(key, 0);
+
+ CacheEntry<Integer, Integer> entry = cache.getEntry(key);
+
+ assertNotNull(entry);
+ assertNotNull(entry.version());
+
+ Ignite ign = cache.unwrap(Ignite.class);
+
+ try (Transaction tx = ign.transactions().txStart(PESSIMISTIC,
txIsolation)) {
+ if (batch)
+ assertTrue(acquireLockForEntries(cache, List.of(entry), 0));
+ else
+ assertTrue(acquireLockForEntry(cache, entry, 0));
+
+ assertEquals(0, cache.get(key).intValue());
+
+ cache.put(key, 1);
+
+ assertEquals(1, cache.get(key).intValue());
+
+ tx.commit();
+ }
+
+ assertEquals(1, cache.get(key).intValue());
+ assertTrue(cache.getEntry(key).version().compareTo(entry.version()) >
0);
+ }
+
+ /**
+ * Acquires a transactional lock for an entry.
+ *
+ * @param cache Cache.
+ * @param entry Entry.
+ * @param timeout Lock wait timeout.
+ * @return {@code true} if the lock was acquired.
+ * @throws IgniteCheckedException If failed.
+ */
+ @SuppressWarnings("unchecked")
+ private static boolean acquireLockForEntry(
+ IgniteCache<Integer, Integer> cache,
+ CacheEntry<Integer, Integer> entry,
+ long timeout
+ ) throws IgniteCheckedException {
+ return
cache.unwrap(IgniteCacheProxy.class).internalProxy().lockTxEntry(entry,
timeout);
+ }
+
+ /**
+ * Acquires transactional locks for entries.
+ *
+ * @param cache Cache.
+ * @param entries Entries.
+ * @param timeout Lock wait timeout.
+ * @return {@code true} if all locks were acquired.
+ * @throws IgniteCheckedException If failed.
+ */
+ @SuppressWarnings("unchecked")
+ private static boolean acquireLockForEntries(
+ IgniteCache<Integer, Integer> cache,
+ List<CacheEntry<Integer, Integer>> entries,
+ long timeout
+ ) throws IgniteCheckedException {
+ return
cache.unwrap(IgniteCacheProxy.class).internalProxy().lockTxEntries(entries,
timeout);
+ }
+}
diff --git
a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/LockTxEntryOneNodeTest.java
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/LockTxEntryOneNodeTest.java
new file mode 100644
index 00000000000..087526b4106
--- /dev/null
+++
b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/LockTxEntryOneNodeTest.java
@@ -0,0 +1,602 @@
+/*
+ * 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.Collection;
+import java.util.List;
+import java.util.concurrent.Callable;
+import javax.cache.processor.EntryProcessor;
+import javax.cache.processor.EntryProcessorException;
+import javax.cache.processor.MutableEntry;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.IgniteCheckedException;
+import org.apache.ignite.cache.CacheAtomicityMode;
+import org.apache.ignite.cache.CacheEntry;
+import org.apache.ignite.cache.CacheMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.NearCacheConfiguration;
+import org.apache.ignite.internal.IgniteInternalFuture;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.apache.ignite.transactions.Transaction;
+import org.apache.ignite.transactions.TransactionTimeoutException;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+import static org.apache.ignite.transactions.TransactionConcurrency.OPTIMISTIC;
+import static
org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC;
+import static
org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED;
+import static
org.apache.ignite.transactions.TransactionIsolation.REPEATABLE_READ;
+
+/**
+ * Tests transactional entry locking through internal cache API.
+ */
+@RunWith(Parameterized.class)
+public class LockTxEntryOneNodeTest extends GridCommonAbstractTest {
+ /** */
+ private static final int KEY = 1;
+
+ /** */
+ private static final int INIT_VAL = 1;
+
+ /** */
+ private static final int UPDATED_VAL = 2;
+
+ /** */
+ private static Ignite ignite;
+
+ /** */
+ @Parameterized.Parameter(0)
+ public boolean commit;
+
+ /** */
+ @Parameterized.Parameter(1)
+ public boolean useNearCache;
+
+ /** */
+ @Parameterized.Parameter(2)
+ public CacheMode cacheMode;
+
+ /** Operation that enlists an entry in a transaction before a versioned
lock is acquired. */
+ private enum TxOperation {
+ /** Read. */
+ GET,
+
+ /** Update. */
+ PUT,
+
+ /** Update with previous value returned. */
+ GET_AND_PUT,
+
+ /** Delete. */
+ REMOVE,
+
+ /** Delete with previous value returned. */
+ GET_AND_REMOVE,
+
+ /** Transform. */
+ INVOKE
+ }
+
+ /** Cache. */
+ private IgniteCache<Integer, Integer> cache;
+
+ /**
+ * Returns data for test.
+ *
+ * @return Test parameters.
+ */
+ @Parameterized.Parameters(name = "commit={0}, useNearCache={1},
cacheMode={2}")
+ public static Collection<Object[]> testData() {
+ return List.of(new Object[][] {
+ {false, false, PARTITIONED},
+ {false, false, REPLICATED},
+ {true, false, PARTITIONED},
+ {true, false, REPLICATED},
+ {false, true, PARTITIONED},
+ {false, true, REPLICATED},
+ {true, true, PARTITIONED},
+ {true, true, REPLICATED},
+ });
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTestsStarted() throws Exception {
+ super.beforeTestsStarted();
+
+ ignite = startGrid(0);
+
+ awaitPartitionMapExchange();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTestsStopped() throws Exception {
+ stopAllGrids();
+
+ ignite = null;
+
+ super.afterTestsStopped();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ cache = ignite.createCache(new CacheConfiguration<Integer,
Integer>(DEFAULT_CACHE_NAME)
+ .setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL)
+ .setNearConfiguration(useNearCache ? new
NearCacheConfiguration<>() : null)
+ .setCacheMode(cacheMode));
+
+ cache.put(KEY, INIT_VAL);
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ ignite.destroyCache(DEFAULT_CACHE_NAME);
+
+ super.afterTest();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected long getTestTimeout() {
+ return 30_000;
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockTxEntryInPessimisticTransaction() throws Exception {
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+
+ try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertTrue(acquireLockForEntry(entry, 0));
+
+ assertEquals(entry.getValue(), cache.get(KEY));
+
+ checkInaccessInOtherTx(cache);
+
+ if (commit)
+ tx.commit();
+ }
+
+ assertEquals(INIT_VAL, cache.get(KEY).intValue());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockAlreadyLockedTxEntry() throws Exception {
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+
+ try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ cache.put(KEY, UPDATED_VAL);
+
+ assertTrue(acquireLockForEntry(entry, 0));
+
+ assertEquals(UPDATED_VAL, cache.get(KEY).intValue());
+
+ checkInaccessInOtherTx(cache);
+
+ if (commit)
+ tx.commit();
+ }
+
+ assertEquals(commit ? UPDATED_VAL : INIT_VAL,
cache.get(KEY).intValue());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockAlreadyEnlistedTxEntryAfterOperation() throws
Exception {
+ for (TxOperation op : TxOperation.values()) {
+ int key = KEY + op.ordinal() + 1;
+
+ cache.put(key, INIT_VAL);
+
+ CacheEntry<Integer, Integer> entry = cache.getEntry(key);
+
+ try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ applyOperation(op, key);
+
+ try {
+ assertTrue("Failed to lock already enlisted tx entry [op="
+ op + ']',
+ acquireLockForEntry(entry, 0));
+ }
+ catch (AssertionError e) {
+ throw new AssertionError("Failed to lock already enlisted
tx entry [op=" + op + ']', e);
+ }
+
+ checkInaccessInOtherTx(cache, key);
+
+ if (commit)
+ tx.commit();
+ }
+
+ assertEquals("Unexpected value after transaction [op=" + op + ']',
+ commit ? expectedValue(op) : Integer.valueOf(INIT_VAL),
+ cache.get(key));
+ }
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testFailToLockTxEntryInPessimisticTransaction() throws
Exception {
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+
+ cache.put(KEY, UPDATED_VAL);
+
+ try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertFalse(acquireLockForEntry(entry, 0));
+
+ assertEquals(UPDATED_VAL, cache.get(KEY).intValue());
+
+ checkAccessInOtherTx(cache);
+
+ if (commit)
+ tx.commit();
+ }
+
+ assertEquals(UPDATED_VAL, cache.get(KEY).intValue());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void
testLockTxEntryReturnsFalseOnTimeoutWhenLockedInOtherTransaction() throws
Exception {
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+
+ try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertTrue(acquireLockForEntry(entry, 0));
+
+ IgniteInternalFuture<Boolean> lockFut = GridTestUtils.runAsync(new
Callable<Boolean>() {
+ @Override public Boolean call() throws Exception {
+ boolean locked = true;
+
+ try (Transaction tx =
ignite.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
+ locked = acquireLockForEntry(entry, 100);
+
+ assertFalse(locked);
+
+ cache.put(2, 2);
+
+ if (commit)
+ tx.commit();
+ }
+
+ return locked;
+ }
+ });
+
+ assertFalse(lockFut.get(10_000));
+
+ if (commit)
+ tx.commit();
+ }
+
+ assertEquals(INIT_VAL, cache.get(KEY).intValue());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockTxEntryDoesNotWaitMuchLongerThanTimeout() throws
Exception {
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+
+ try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertTrue(acquireLockForEntry(entry, 0));
+
+ IgniteInternalFuture<Long> lockFut = GridTestUtils.runAsync(new
Callable<Long>() {
+ @Override public Long call() throws Exception {
+ long waitTime;
+
+ try (Transaction tx =
ignite.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
+ long start = U.currentTimeMillis();
+
+ assertFalse(acquireLockForEntry(entry, 200));
+
+ waitTime = U.currentTimeMillis() - start;
+
+ if (commit)
+ tx.commit();
+ }
+
+ return waitTime;
+ }
+ });
+
+ long waitTime = lockFut.get(10_000);
+
+ assertTrue("Waited for lock for " + waitTime + " ms it is
unexpectedly long.",
+ waitTime < 1_000);
+
+ if (commit)
+ tx.commit();
+ }
+
+ assertEquals(INIT_VAL, cache.get(KEY).intValue());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockTxEntryReturnsFalseForLockedKeyAndTrueForOtherKey()
throws Exception {
+ int otherKey = KEY + 1;
+
+ cache.put(otherKey, INIT_VAL);
+
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+ CacheEntry<Integer, Integer> otherEntry = cache.getEntry(otherKey);
+
+ try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertTrue(acquireLockForEntry(entry, 0));
+
+ IgniteInternalFuture<Void> lockFut = GridTestUtils.runAsync(new
Callable<Void>() {
+ @Override public Void call() throws Exception {
+ try (Transaction tx =
ignite.transactions().txStart(PESSIMISTIC, READ_COMMITTED)) {
+ assertFalse(acquireLockForEntry(entry, -1));
+ assertTrue(acquireLockForEntry(otherEntry, -1));
+
+ if (commit)
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ lockFut.get(10_000);
+
+ if (commit)
+ tx.commit();
+ }
+
+ assertEquals(INIT_VAL, cache.get(KEY).intValue());
+ assertEquals(INIT_VAL, cache.get(otherKey).intValue());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testUpdateAfterLock() throws Exception {
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+
+ try (Transaction tx = ignite.transactions().txStart(PESSIMISTIC,
READ_COMMITTED)) {
+ assertTrue(acquireLockForEntry(entry, 0));
+
+ checkInaccessInOtherTx(cache);
+
+ assertEquals(entry.getValue(), cache.get(KEY));
+
+ cache.put(KEY, UPDATED_VAL);
+
+ assertEquals(UPDATED_VAL, cache.get(KEY).intValue());
+
+ if (commit)
+ tx.commit();
+ }
+
+ assertEquals(commit ? UPDATED_VAL : INIT_VAL,
cache.get(KEY).intValue());
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockTxEntryFailsWithoutTransaction() throws Exception {
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+
+ GridTestUtils.assertThrows(log, new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ acquireLockForEntry(entry, 0);
+
+ return null;
+ }
+ }, IgniteCheckedException.class, "without transaction");
+ }
+
+ /**
+ * @throws Exception If failed.
+ */
+ @Test
+ public void testLockTxEntryAsyncFailsInOptimisticTransaction() throws
Exception {
+ CacheEntry<Integer, Integer> entry = cache.getEntry(KEY);
+
+ try (Transaction tx = ignite.transactions().txStart(OPTIMISTIC,
REPEATABLE_READ)) {
+ GridTestUtils.assertThrows(log, new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ return acquireLockForEntry(entry, 0);
+ }
+ }, IgniteCheckedException.class, "optimistic transaction");
+ }
+ }
+
+ /**
+ * Checks that another transaction can access the cache.
+ *
+ * @param cache Cache.
+ * @throws IgniteCheckedException If failed.
+ */
+ private void checkAccessInOtherTx(IgniteCache<Integer, Integer> cache)
throws IgniteCheckedException {
+ checkAccessInOtherTx(cache, KEY);
+ }
+
+ /**
+ * Checks that another transaction can access the cache.
+ *
+ * @param cache Cache.
+ * @param key Key.
+ * @throws IgniteCheckedException If failed.
+ */
+ private void checkAccessInOtherTx(IgniteCache<Integer, Integer> cache, int
key) throws IgniteCheckedException {
+ IgniteInternalFuture<Void> accessFut = GridTestUtils.runAsync(new
Callable<Void>() {
+ @Override public Void call() {
+ try (Transaction tx =
ignite.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 100, 1)) {
+ cache.get(key);
+
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ accessFut.get(10_000);
+ }
+
+ /**
+ * Checks that another transaction cannot access the cache.
+ *
+ * @param cache Cache.
+ * @throws IgniteCheckedException If failed.
+ */
+ private void checkInaccessInOtherTx(IgniteCache<Integer, Integer> cache)
throws IgniteCheckedException {
+ checkInaccessInOtherTx(cache, KEY);
+ }
+
+ /**
+ * Checks that another transaction cannot access the cache.
+ *
+ * @param cache Cache.
+ * @param key Key.
+ * @throws IgniteCheckedException If failed.
+ */
+ private void checkInaccessInOtherTx(IgniteCache<Integer, Integer> cache,
int key) throws IgniteCheckedException {
+ IgniteInternalFuture<Void> accessFut = GridTestUtils.runAsync(new
Callable<Void>() {
+ @Override public Void call() {
+ try (Transaction tx =
ignite.transactions().txStart(PESSIMISTIC, REPEATABLE_READ, 100, 1)) {
+ cache.get(key);
+
+ tx.commit();
+ }
+
+ return null;
+ }
+ });
+
+ GridTestUtils.assertThrowsWithCause(new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ return accessFut.get(10_000);
+ }
+ }, TransactionTimeoutException.class);
+ }
+
+ /**
+ * Acquires a transactional lock for an entry.
+ *
+ * @param entry Entry.
+ * @param timeout Lock wait timeout.
+ * @return {@code true} if the lock was acquired.
+ * @throws IgniteCheckedException If failed.
+ */
+ @SuppressWarnings("unchecked")
+ private boolean acquireLockForEntry(
+ CacheEntry<Integer, Integer> entry,
+ long timeout
+ ) throws IgniteCheckedException {
+ return
cache.unwrap(IgniteCacheProxy.class).internalProxy().lockTxEntry(entry,
timeout);
+ }
+
+ /**
+ * Applies cache operation inside transaction.
+ *
+ * @param op Operation.
+ * @param key Key.
+ */
+ private void applyOperation(TxOperation op, int key) {
+ switch (op) {
+ case GET:
+ assertEquals(INIT_VAL, cache.get(key).intValue());
+
+ break;
+
+ case PUT:
+ cache.put(key, UPDATED_VAL);
+
+ break;
+
+ case GET_AND_PUT:
+ assertEquals(INIT_VAL, cache.getAndPut(key,
UPDATED_VAL).intValue());
+
+ break;
+
+ case REMOVE:
+ assertTrue(cache.remove(key));
+
+ break;
+
+ case GET_AND_REMOVE:
+ assertEquals(INIT_VAL, cache.getAndRemove(key).intValue());
+
+ break;
+
+ case INVOKE:
+ cache.invoke(key, new EntryProcessor<Integer, Integer,
Object>() {
+ @Override public Object process(MutableEntry<Integer,
Integer> entry, Object... args)
+ throws EntryProcessorException {
+ entry.setValue(UPDATED_VAL);
+
+ return null;
+ }
+ });
+
+ break;
+
+ default:
+ fail("Unexpected operation: " + op);
+ }
+ }
+
+ /**
+ * @param op Operation.
+ * @return Expected value after committed transaction.
+ */
+ private Integer expectedValue(TxOperation op) {
+ switch (op) {
+ case GET:
+ return INIT_VAL;
+
+ case PUT:
+ case GET_AND_PUT:
+ case INVOKE:
+ return UPDATED_VAL;
+
+ case REMOVE:
+ case GET_AND_REMOVE:
+ return null;
+
+ default:
+ fail("Unexpected operation: " + op);
+
+ return null;
+ }
+ }
+}
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 bef893765b2..5cd49400af7 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
@@ -28,12 +28,14 @@ import
org.apache.ignite.internal.processors.cache.CacheGetEntryOptimisticSerial
import
org.apache.ignite.internal.processors.cache.CacheGetEntryPessimisticReadCommittedSelfTest;
import
org.apache.ignite.internal.processors.cache.CacheGetEntryPessimisticSerializableSelfTest;
import
org.apache.ignite.internal.processors.cache.CacheOffheapMapEntrySelfTest;
+import
org.apache.ignite.internal.processors.cache.CacheOperationContextTransactionalLockTest;
import
org.apache.ignite.internal.processors.cache.CacheReadThroughReplicatedAtomicRestartSelfTest;
import
org.apache.ignite.internal.processors.cache.CacheReadThroughReplicatedRestartSelfTest;
import
org.apache.ignite.internal.processors.cache.CacheReadThroughRestartSelfTest;
import
org.apache.ignite.internal.processors.cache.CacheStoreUsageMultinodeDynamicStartTxTest;
import
org.apache.ignite.internal.processors.cache.CacheStoreUsageMultinodeStaticStartTxTest;
import
org.apache.ignite.internal.processors.cache.CacheTxNotAllowReadFromBackupTest;
+import
org.apache.ignite.internal.processors.cache.CacheVersionedEntryTransactionalLockTest;
import
org.apache.ignite.internal.processors.cache.GridCacheMultinodeUpdateAtomicNearEnabledSelfTest;
import
org.apache.ignite.internal.processors.cache.GridCacheMultinodeUpdateAtomicSelfTest;
import
org.apache.ignite.internal.processors.cache.GridCacheMultinodeUpdateNearEnabledNoBackupsSelfTest;
@@ -58,6 +60,7 @@ import
org.apache.ignite.internal.processors.cache.IgniteDynamicCacheWithConfigS
import
org.apache.ignite.internal.processors.cache.IgniteExchangeFutureHistoryTest;
import
org.apache.ignite.internal.processors.cache.IgniteStartCacheInTransactionSelfTest;
import
org.apache.ignite.internal.processors.cache.IgniteSystemCacheOnClientTest;
+import org.apache.ignite.internal.processors.cache.LockTxEntryOneNodeTest;
import
org.apache.ignite.internal.processors.cache.MarshallerCacheJobRunNodeRestartTest;
import
org.apache.ignite.internal.processors.cache.distributed.CacheDirectoryNameTest;
import
org.apache.ignite.internal.processors.cache.distributed.CacheDiscoveryDataConcurrentJoinTest;
@@ -162,6 +165,10 @@ public class IgniteCacheTestSuite4 {
GridTestUtils.addTestIfNeeded(suite,
CacheReadThroughReplicatedAtomicRestartSelfTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
CacheVersionedEntryPartitionedAtomicSelfTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
CacheVersionedEntryReplicatedTransactionalSelfTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite,
CacheVersionedEntryTransactionalLockTest.class, ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite, LockTxEntryOneNodeTest.class,
ignoredTests);
+ GridTestUtils.addTestIfNeeded(suite,
CacheOperationContextTransactionalLockTest.class, ignoredTests);
+
GridTestUtils.addTestIfNeeded(suite,
GridCacheDhtTxPreloadSelfTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite,
GridCacheNearTxPreloadSelfTest.class, ignoredTests);
GridTestUtils.addTestIfNeeded(suite, CacheGroupsPreloadTest.class,
ignoredTests);
