From 65130e718ca5e54eff0ec19f063f50ee1006a30f Mon Sep 17 00:00:00 2001 From: Vladislav Pyatkov Date: Wed, 11 Nov 2026 14:47:35 +0300 Subject: [PATCH 1/2] IGNITE-29279 Enforce transactional lock wait timeouts on primary nodes --- .../colocated/GridDhtColocatedLockFuture.java | 53 +++--- .../distributed/near/GridNearLockFuture.java | 54 +++--- .../distributed/near/GridNearLockRequest.java | 7 + .../distributed/near/GridNearTxLocal.java | 8 +- .../cache/CacheLockWaitTimeoutTest.java | 180 ++++++++++++++++++ .../testsuites/IgniteCacheTestSuite4.java | 3 + 6 files changed, 243 insertions(+), 62 deletions(-) create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheLockWaitTimeoutTest.java diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java index 1dc33ad688730..011c8933f4ed2 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 @@ -140,6 +140,9 @@ public final class GridDhtColocatedLockFuture extends GridCacheCompoundIdentityF /** Lock wait timeout. */ private final long waitTimeout; + /** Time when the first primary batch starts, or zero before mapping. */ + private long lockWaitStartTime; + /** Transaction. */ @GridToStringExclude private final GridNearTxLocal tx; @@ -805,7 +808,7 @@ void map() { if (isDone()) // Possible due to async rollback. return; - if (lockTimeout() > 0) { + if (timeout > 0) { timeoutObj = new LockTimeoutObject(); cctx.time().addTimeoutObject(timeoutObj); @@ -1250,9 +1253,24 @@ private void proceedMapping0() final Collection mappedKeys = map.distributedKeys(); final ClusterNode node = map.node(); + long remainingWaitTimeout = waitTimeout; + + if (waitTimeout > 0) { + long now = U.currentTimeMillis(); + + if (lockWaitStartTime == 0) + lockWaitStartTime = now; + + long elapsed = now - lockWaitStartTime; + + remainingWaitTimeout = elapsed < waitTimeout ? waitTimeout - elapsed : -1; + } + if (node.isLocal()) - lockLocally(mappedKeys, req.topologyVersion()); + lockLocally(mappedKeys, req.topologyVersion(), remainingWaitTimeout); else { + req.waitTimeout(remainingWaitTimeout); + final MiniFuture fut = new MiniFuture(node, mappedKeys, ++miniId); req.miniId(fut.futureId()); @@ -1280,10 +1298,12 @@ private void proceedMapping0() * Locks given keys directly through dht cache. * @param keys Collection of keys. * @param topVer Topology version to lock on. + * @param remainingWaitTimeout Remaining lock wait budget for this batch. */ private void lockLocally( final Collection keys, - AffinityTopologyVersion topVer + AffinityTopologyVersion topVer, + long remainingWaitTimeout ) { if (log.isDebugEnabled()) log.debug("Before locally locking keys : " + keys); @@ -1297,7 +1317,7 @@ private void lockLocally( read, retval, timeout, - waitTimeout, + remainingWaitTimeout, createTtl, accessTtl, skipStore, @@ -1418,7 +1438,7 @@ private boolean mapAsPrimary(Collection keys, AffinityTopologyVe tx.addKeyMapping(cctx.txKey(key), cctx.localNode()); } - lockLocally(distributedKeys, topVer); + lockLocally(distributedKeys, topVer, waitTimeout); } GridDhtPartitionsExchangeFuture lastFinishedFut = cctx.shared().exchange().lastFinishedFuture(); @@ -1525,13 +1545,6 @@ private boolean errorOrTimeoutOnTopologyVersion(IgniteCheckedException e, boolea return false; } - /** - * @return Timeout value for this lock future. - */ - private long lockTimeout() { - return CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout) ? waitTimeout : timeout; - } - /** * Lock request timeout object. */ @@ -1540,7 +1553,7 @@ private class LockTimeoutObject extends GridTimeoutObjectAdapter { * Default constructor. */ LockTimeoutObject() { - super(lockTimeout()); + super(timeout); } /** Requested keys. */ @@ -1551,20 +1564,6 @@ private class LockTimeoutObject extends GridTimeoutObjectAdapter { 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) { 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 c5eb21e6b3147..b9748104b7e61 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 @@ -137,6 +137,9 @@ public final class GridNearLockFuture extends GridCacheCompoundIdentityFuture 0) { + if (timeout > 0) { timeoutObj = new LockTimeoutObject(); cctx.time().addTimeoutObject(timeoutObj); @@ -1206,6 +1204,21 @@ private void proceedMapping0() final Collection mappedKeys = map.distributedKeys(); final ClusterNode node = map.node(); + long remainingWaitTimeout = waitTimeout; + + if (waitTimeout > 0) { + long now = U.currentTimeMillis(); + + if (lockWaitStartTime == 0) + lockWaitStartTime = now; + + long elapsed = now - lockWaitStartTime; + + remainingWaitTimeout = elapsed < waitTimeout ? waitTimeout - elapsed : -1; + } + + req.waitTimeout(remainingWaitTimeout); + if (node.isLocal()) { req.miniId(-1); @@ -1420,13 +1433,6 @@ private ClusterTopologyCheckedException newTopologyException(@Nullable Throwable return topEx; } - /** - * @return Timeout value for this lock future. - */ - private long lockTimeout() { - return CU.isWaitTimeoutExpiresFirst(waitTimeout, timeout) ? waitTimeout : timeout; - } - /** * Lock request timeout object. */ @@ -1435,7 +1441,7 @@ private class LockTimeoutObject extends GridTimeoutObjectAdapter { * Default constructor. */ LockTimeoutObject() { - super(lockTimeout()); + super(timeout); } /** Requested keys. */ @@ -1448,20 +1454,6 @@ private class LockTimeoutObject extends GridTimeoutObjectAdapter { 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()) { 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 46fae249ebd1f..02c41d9b5498c 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 @@ -221,6 +221,13 @@ public long waitTimeout() { return waitTimeout; } + /** + * @param timeout Remaining lock wait budget for this primary batch. + */ + public void waitTimeout(long timeout) { + waitTimeout = timeout; + } + /** * @return Topology version. */ 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 36630c33fb49c..9a99fe2ec3fc9 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 @@ -647,7 +647,7 @@ private IgniteInternalFuture putAsync0( IgniteInternalFuture fut = cacheCtx.cache().txLockAsync(enlisted, timeout, - timeout, + /*waitTimeout*/0, this, /*read*/entryProc != null, // Needed to force load from store. retval, @@ -822,7 +822,7 @@ private IgniteInternalFuture putAllAsync0( IgniteInternalFuture fut = cacheCtx.cache().txLockAsync(enlisted, timeout, - timeout, + /*waitTimeout*/0, this, /*read*/invokeVals != null, // Needed to force load from store. retval, @@ -1725,7 +1725,7 @@ private IgniteInternalFuture removeAllAsync0( IgniteInternalFuture fut = cacheCtx.cache().txLockAsync(enlisted, timeout, - timeout, + /*waitTimeout*/0, this, false, retval, @@ -1918,7 +1918,7 @@ public IgniteInternalFuture> getAllAsync( IgniteInternalFuture fut = cacheCtx.cache().txLockAsync(lockKeys, timeout, - timeout, + /*waitTimeout*/0, this, true, true, diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheLockWaitTimeoutTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheLockWaitTimeoutTest.java new file mode 100644 index 0000000000000..953b627f7116d --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheLockWaitTimeoutTest.java @@ -0,0 +1,180 @@ +/* + * 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.IgniteCache; +import org.apache.ignite.cache.CacheEntry; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.configuration.NearCacheConfiguration; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.TestRecordingCommunicationSpi; +import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockRequest; +import org.apache.ignite.internal.processors.cache.distributed.near.GridNearLockResponse; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.apache.ignite.transactions.Transaction; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL; +import static org.apache.ignite.cache.CacheWriteSynchronizationMode.FULL_SYNC; +import static org.apache.ignite.internal.TestRecordingCommunicationSpi.spi; +import static org.apache.ignite.transactions.TransactionConcurrency.PESSIMISTIC; +import static org.apache.ignite.transactions.TransactionIsolation.READ_COMMITTED; + +/** Tests primary-side lock wait deadlines independently of response delivery. */ +@RunWith(Parameterized.class) +public class CacheLockWaitTimeoutTest extends GridCommonAbstractTest { + /** Lock wait budget, shorter than the transaction timeout. */ + private static final long WAIT_TIMEOUT = 500; + + /** Whether the initiating client uses a near cache. */ + @Parameterized.Parameter + public boolean near; + + /** @return Cache configurations. */ + @Parameterized.Parameters(name = "near={0}") + public static Collection parameters() { + return List.of(new Object[] {false}, new Object[] {true}); + } + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String name) throws Exception { + return super.getConfiguration(name).setCommunicationSpi(new TestRecordingCommunicationSpi()); + } + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + startGridsMultiThreaded(2); + startClientGrid(2); + } + + /** {@inheritDoc} */ + @Override protected void beforeTest() throws Exception { + grid(0).createCache(new CacheConfiguration(DEFAULT_CACHE_NAME) + .setAtomicityMode(TRANSACTIONAL) + .setBackups(1) + .setWriteSynchronizationMode(FULL_SYNC)); + + if (near) + grid(2).createNearCache(DEFAULT_CACHE_NAME, new NearCacheConfiguration<>()); + else + grid(2).cache(DEFAULT_CACHE_NAME); + + awaitPartitionMapExchange(); + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + TestRecordingCommunicationSpi.stopBlockAll(); + grid(0).destroyCache(DEFAULT_CACHE_NAME); + } + + /** {@inheritDoc} */ + @Override protected void afterTestsStopped() throws Exception { + stopAllGrids(); + } + + /** A successful primary response remains valid after the coordinator's lock wait budget expires. */ + @Test + public void testDelayedSuccessfulResponse() throws Exception { + IgniteCache cache = grid(2).cache(DEFAULT_CACHE_NAME); + int key = primaryKey(grid(0).cache(DEFAULT_CACHE_NAME)); + + cache.put(key, 0); + + CacheEntry entry = cache.getEntry(key); + + spi(grid(0)).blockMessages(GridNearLockResponse.class, grid(2).name()); + + IgniteInternalFuture fut = lockAsync(cache, List.of(entry)); + + try { + assertTrue(spi(grid(0)).waitForBlocked(1, 10_000)); + assertFalse("The coordinator must await the primary response", + GridTestUtils.waitForCondition(fut::isDone, 2 * WAIT_TIMEOUT)); + } + finally { + spi(grid(0)).stopBlock(); + } + + assertTrue(fut.get(10_000)); + } + + /** Later primary batches must not receive a fresh lock wait budget. */ + @Test + public void testWaitBudgetIsSharedByPrimaryBatches() throws Exception { + IgniteCache cache = grid(2).cache(DEFAULT_CACHE_NAME); + IgniteCache holderCache = grid(1).cache(DEFAULT_CACHE_NAME); + int first = primaryKey(grid(0).cache(DEFAULT_CACHE_NAME)); + int second = primaryKey(holderCache); + + cache.put(first, 0); + cache.put(second, 0); + + List> entries = List.of(cache.getEntry(first), cache.getEntry(second)); + + try (Transaction holder = grid(1).transactions().txStart(PESSIMISTIC, READ_COMMITTED)) { + holderCache.put(second, 1); + + spi(grid(2)).record(GridNearLockRequest.class); + spi(grid(0)).blockMessages(GridNearLockResponse.class, grid(2).name()); + + IgniteInternalFuture fut = lockAsync(cache, entries); + + try { + assertTrue(spi(grid(0)).waitForBlocked(1, 10_000)); + assertFalse("The coordinator must await the primary response", + GridTestUtils.waitForCondition(fut::isDone, 2 * WAIT_TIMEOUT)); + } + finally { + spi(grid(0)).stopBlock(); + } + + assertFalse(fut.get(10_000)); + + List requests = spi(grid(2)).recordedMessages(true); + + assertEquals(2, requests.size()); + assertTrue(((GridNearLockRequest)requests.get(0)).waitTimeout() > 0); + assertTrue("An exhausted budget requires an immediate attempt on the next primary", + ((GridNearLockRequest)requests.get(1)).waitTimeout() < 0); + } + } + + /** + * @param cache Client cache. + * @param entries Entries to lock, in primary mapping order. + * @return Lock result after transaction cleanup. + */ + @SuppressWarnings("unchecked") + private IgniteInternalFuture lockAsync( + IgniteCache cache, + List> entries + ) { + return GridTestUtils.runAsync(() -> { + try (Transaction tx = grid(2).transactions().txStart(PESSIMISTIC, READ_COMMITTED, 30_000, 0)) { + return cache.unwrap(IgniteCacheProxy.class).internalProxy().lockTxEntries(entries, WAIT_TIMEOUT); + } + }); + } +} 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 5cd49400af78b..e5a4c64475878 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 @@ -27,6 +27,7 @@ import org.apache.ignite.internal.processors.cache.CacheGetEntryOptimisticSerializableSelfTest; import org.apache.ignite.internal.processors.cache.CacheGetEntryPessimisticReadCommittedSelfTest; import org.apache.ignite.internal.processors.cache.CacheGetEntryPessimisticSerializableSelfTest; +import org.apache.ignite.internal.processors.cache.CacheLockWaitTimeoutTest; import org.apache.ignite.internal.processors.cache.CacheOffheapMapEntrySelfTest; import org.apache.ignite.internal.processors.cache.CacheOperationContextTransactionalLockTest; import org.apache.ignite.internal.processors.cache.CacheReadThroughReplicatedAtomicRestartSelfTest; @@ -165,6 +166,8 @@ public static List> suite(Collection ignoredTests) { GridTestUtils.addTestIfNeeded(suite, CacheReadThroughReplicatedAtomicRestartSelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, CacheVersionedEntryPartitionedAtomicSelfTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, CacheVersionedEntryReplicatedTransactionalSelfTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, CacheLockWaitTimeoutTest.class, ignoredTests); + GridTestUtils.addTestIfNeeded(suite, CacheVersionedEntryTransactionalLockTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, LockTxEntryOneNodeTest.class, ignoredTests); GridTestUtils.addTestIfNeeded(suite, CacheOperationContextTransactionalLockTest.class, ignoredTests); From 8f4eab8375f08c776c8fdde3a487ecb9b22dc07d Mon Sep 17 00:00:00 2001 From: Vladislav Pyatkov Date: Wed, 11 Nov 2026 17:01:18 +0300 Subject: [PATCH 2/2] Optimize tests and remove redundant waitTimeout initialization from the lock request constructor. --- .../colocated/GridDhtColocatedLockFuture.java | 1 - .../distributed/near/GridNearLockFuture.java | 1 - .../distributed/near/GridNearLockRequest.java | 3 --- .../cache/CacheLockWaitTimeoutTest.java | 21 ++++++++++++------- .../ignite/testframework/GridTestUtils.java | 11 ++++++++++ 5 files changed, 25 insertions(+), 12 deletions(-) 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 011c8933f4ed2..ff611bd669965 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 @@ -1127,7 +1127,6 @@ private synchronized void map0( isolation(), isInvalidate(), timeout, - waitTimeout, mappedKeys.size(), inTx() ? tx.size() : mappedKeys.size(), inTx() && tx.syncMode() == FULL_SYNC, 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 b9748104b7e61..73e780ed6a9e6 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 @@ -1086,7 +1086,6 @@ private void map(Iterable keys, boolean remap, boolean topLocked isolation(), isInvalidate(), timeout, - waitTimeout, mappedKeys.size(), inTx() ? tx.size() : mappedKeys.size(), inTx() && tx.syncMode() == FULL_SYNC, 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 02c41d9b5498c..401c014cf26b3 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 @@ -103,7 +103,6 @@ public GridNearLockRequest() { * @param isolation Transaction isolation. * @param isInvalidate Invalidation flag. * @param timeout Transaction timeout. - * @param waitTimeout Lock wait timeout. * @param keyCnt Number of keys. * @param txSize Expected transaction size. * @param syncCommit Synchronous commit flag. @@ -128,7 +127,6 @@ public GridNearLockRequest( TransactionIsolation isolation, boolean isInvalidate, long timeout, - long waitTimeout, int keyCnt, int txSize, boolean syncCommit, @@ -168,7 +166,6 @@ public GridNearLockRequest( this.taskNameHash = taskNameHash; this.createTtl = createTtl; this.accessTtl = accessTtl; - this.waitTimeout = waitTimeout; this.txLbl = txLbl; diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheLockWaitTimeoutTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheLockWaitTimeoutTest.java index 953b627f7116d..76faeb382303c 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheLockWaitTimeoutTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/CacheLockWaitTimeoutTest.java @@ -110,8 +110,7 @@ public void testDelayedSuccessfulResponse() throws Exception { try { assertTrue(spi(grid(0)).waitForBlocked(1, 10_000)); - assertFalse("The coordinator must await the primary response", - GridTestUtils.waitForCondition(fut::isDone, 2 * WAIT_TIMEOUT)); + assertFalse(fut.isDone()); } finally { spi(grid(0)).stopBlock(); @@ -143,8 +142,11 @@ public void testWaitBudgetIsSharedByPrimaryBatches() throws Exception { try { assertTrue(spi(grid(0)).waitForBlocked(1, 10_000)); - assertFalse("The coordinator must await the primary response", - GridTestUtils.waitForCondition(fut::isDone, 2 * WAIT_TIMEOUT)); + + // Send the second batch on a later clock tick so that its remaining wait budget is smaller. + GridTestUtils.waitForNextClockTick(); + + assertFalse(fut.isDone()); } finally { spi(grid(0)).stopBlock(); @@ -155,9 +157,14 @@ public void testWaitBudgetIsSharedByPrimaryBatches() throws Exception { List requests = spi(grid(2)).recordedMessages(true); assertEquals(2, requests.size()); - assertTrue(((GridNearLockRequest)requests.get(0)).waitTimeout() > 0); - assertTrue("An exhausted budget requires an immediate attempt on the next primary", - ((GridNearLockRequest)requests.get(1)).waitTimeout() < 0); + + long firstWaitTimeout = ((GridNearLockRequest)requests.get(0)).waitTimeout(); + long secondWaitTimeout = ((GridNearLockRequest)requests.get(1)).waitTimeout(); + + assertTrue(firstWaitTimeout > 0); + assertTrue("The next primary must receive a smaller lock wait budget [first=" + firstWaitTimeout + + ", second=" + secondWaitTimeout + ']', + secondWaitTimeout < firstWaitTimeout); } } diff --git a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java index 3b428455a09ab..8e0ea33a39648 100644 --- a/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java +++ b/modules/core/src/test/java/org/apache/ignite/testframework/GridTestUtils.java @@ -1986,6 +1986,17 @@ public static int sleepAndIncrement(int sleepDur, int i) throws IgniteInterrupte return i + 1; } + /** + * Waits until the cached Ignite clock advances beyond its current value. + * + * @throws IgniteInterruptedCheckedException If interrupted. + */ + public static void waitForNextClockTick() throws IgniteInterruptedCheckedException { + long now = U.currentTimeMillis(); + + assertTrue("Ignite clock did not advance", waitForCondition(() -> U.currentTimeMillis() > now, 1_000, 1)); + } + /** * Waits for condition, polling in busy wait loop. *