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 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 8f453ca4a876e..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 @@ -56,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; @@ -146,6 +147,32 @@ public void testContinuousQueryBuffersCleanup() throws Exception { } } + /** Checks that a backup acknowledgement covering several topology versions is sent to all backups. */ + @Test + public void testBackupAcknowledgeCoveringSeveralTopologyVersions() throws Exception { + startGrids(3); + + // Keys of the first node are in one partition, so backup CQ buffer exists before the acknowledgement. + IgniteCache 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. + 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++) { + 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);