IGNITE-29232 Fix continuous query backup ack not sent to all backups

modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryHandler.java

@@ -1543,7 +1543,7 @@ private void sendBackupAcknowledge(final IgniteBiTuple<Map<Integer, Long>, Set<A
                             for (ClusterNode node : ctx.discovery().cacheGroupAffinityNodes(cctx.groupId(), topVer)) {
                                 if (!node.isLocal()) {
                                     // Try to send the same message only once per node.
-                                    if (sentToNodes != null && !sentToNodes.add(nodeId))
+                                    if (sentToNodes != null && !sentToNodes.add(node.id()))
                                         continue;
 
                                     try {

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();
+    }
 }

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<Object, Object> 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<Query<?>, QueryCursor<?>> queryExecutor) throws Exception {
         IgniteEx primary = grid(primaryNodeIdx);