From 11ff664640ae46d9bfdc2a9523b5a3dc8724ca71 Mon Sep 17 00:00:00 2001 From: Dmitry Werner Date: Fri, 30 Oct 2026 13:15:21 +0100 Subject: [PATCH 1/2] IGNITE-29232 Fix continuous query backup ack sent to single node --- .../CacheContinuousQueryHandler.java | 2 +- .../ContinuousQueryBuffersCleanupTest.java | 29 +++++++++++++++++++ 2 files changed, 30 insertions(+), 1 deletion(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java index b42c353ceb8fa..204f9f76eb35f 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java @@ -1543,7 +1543,7 @@ private void sendBackupAcknowledge(final IgniteBiTuple, Set cache = grid(0).createCache(new CacheConfiguration<>(DEFAULT_CACHE_NAME) + .setAffinity(new GridCacheModuloAffinityFunction(3, 2)) + .setWriteSynchronizationMode(FULL_SYNC) + .setBackups(2)); + + // Disable acknowledgement by timeout, so the single acknowledgement covers both topology versions. + GridTestUtils.getFieldValue(grid(0).cachex(DEFAULT_CACHE_NAME).context().continuousQueries(), + "cancelableTask").close(); + + try (QueryCursor ignored = cache.query(new ContinuousQuery<>().setLocalListener(evts -> {}))) { + for (int i = 0; i < DFLT_CONTINUOUS_QUERY_BACKUP_ACK_THRESHOLD; i++) { + if (i == DFLT_CONTINUOUS_QUERY_BACKUP_ACK_THRESHOLD / 2) + startClientGrid(3); + + cache.put(keyForNode(grid(0)), i); + } + + assertTrue(waitForCondition(() -> G.allGrids().stream().allMatch(this::isContinuesQueryBufferEmpty), 5_000)); + } + } + /** */ private void checkBuffersCleared(int primaryNodeIdx, Function, QueryCursor> queryExecutor) throws Exception { IgniteEx primary = grid(primaryNodeIdx); From 5963ede93f53f774be53fe5b11b92f684f1d788f Mon Sep 17 00:00:00 2001 From: Dmitry Werner Date: Tue, 3 Nov 2026 19:34:40 +0100 Subject: [PATCH 2/2] minor --- .../query/continuous/TestCacheContinuousQueryUtils.java | 6 ++++++ .../continuous/ContinuousQueryBuffersCleanupTest.java | 6 ++---- 2 files changed, 8 insertions(+), 4 deletions(-) diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/TestCacheContinuousQueryUtils.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/TestCacheContinuousQueryUtils.java index 61d1bfba4ca9a..a4389d01609cf 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/TestCacheContinuousQueryUtils.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/TestCacheContinuousQueryUtils.java @@ -18,6 +18,7 @@ package org.apache.ignite.internal.processors.cache.query.continuous; import java.util.Map; +import org.apache.ignite.internal.processors.timeout.GridTimeoutProcessor; import org.apache.ignite.internal.util.GridAtomicLong; /** */ @@ -43,4 +44,9 @@ public static int backupQueueSize(CacheContinuousQueryEventBuffer buf) { public static GridAtomicLong maxReceivedBackupAcknowledgeUpdateCounter(CacheContinuousQueryEventBuffer buf) { return buf.maxReceivedBackupAckUpdCntr; } + + /** */ + public static GridTimeoutProcessor.CancelableTask cancelableTask(CacheContinuousQueryManager mgr) { + return mgr.getCancelableTask(); + } } diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/ContinuousQueryBuffersCleanupTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/ContinuousQueryBuffersCleanupTest.java index 5d19b307f1375..c3dfbb6a70f10 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/ContinuousQueryBuffersCleanupTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/continuous/ContinuousQueryBuffersCleanupTest.java @@ -17,7 +17,6 @@ package org.apache.ignite.internal.processors.continuous; -import java.io.Closeable; import java.util.ArrayList; import java.util.Collection; import java.util.Map; @@ -46,7 +45,6 @@ import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler; import org.apache.ignite.internal.util.typedef.F; import org.apache.ignite.internal.util.typedef.G; -import org.apache.ignite.testframework.GridTestUtils; import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; import org.junit.Test; @@ -58,6 +56,7 @@ import static org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryHandler.DFLT_CONTINUOUS_QUERY_BACKUP_ACK_THRESHOLD; import static org.apache.ignite.internal.processors.cache.query.continuous.TestCacheContinuousQueryUtils.backupQueueSize; import static org.apache.ignite.internal.processors.cache.query.continuous.TestCacheContinuousQueryUtils.bufferedEntries; +import static org.apache.ignite.internal.processors.cache.query.continuous.TestCacheContinuousQueryUtils.cancelableTask; import static org.apache.ignite.internal.processors.cache.query.continuous.TestCacheContinuousQueryUtils.maxReceivedBackupAcknowledgeUpdateCounter; import static org.apache.ignite.internal.processors.cache.query.continuous.TestCacheContinuousQueryUtils.partitionContinuesQueryEntryBuffers; import static org.apache.ignite.testframework.GridTestUtils.waitForCondition; @@ -160,8 +159,7 @@ public void testBackupAcknowledgeCoveringSeveralTopologyVersions() throws Except .setBackups(2)); // Disable acknowledgement by timeout, so the single acknowledgement covers both topology versions. - GridTestUtils.getFieldValue(grid(0).cachex(DEFAULT_CACHE_NAME).context().continuousQueries(), - "cancelableTask").close(); + cancelableTask(grid(0).cachex(DEFAULT_CACHE_NAME).context().continuousQueries()).close(); try (QueryCursor ignored = cache.query(new ContinuousQuery<>().setLocalListener(evts -> {}))) { for (int i = 0; i < DFLT_CONTINUOUS_QUERY_BACKUP_ACK_THRESHOLD; i++) {