From 9d894b6fc0cc21bc140a3fbd14398b3fab69292c Mon Sep 17 00:00:00 2001 From: Reza Taroosheh Date: Wed, 28 Oct 2026 23:19:55 +0100 Subject: [PATCH 1/4] IGNITE-29227 Add a test for transformed events with an unregistered result class on an ATOMIC cache --- ...sWithTransformerUnregisteredClassTest.java | 135 ++++++++++++++++++ .../IgniteCacheQuerySelfTestSuite6.java | 2 + 2 files changed, 137 insertions(+) create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerUnregisteredClassTest.java diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerUnregisteredClassTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerUnregisteredClassTest.java new file mode 100644 index 0000000000000..b346cdef337eb --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerUnregisteredClassTest.java @@ -0,0 +1,135 @@ +/* + * 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.query.continuous; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import javax.cache.configuration.FactoryBuilder; +import javax.cache.event.CacheEntryEvent; +import org.apache.ignite.IgniteCache; +import org.apache.ignite.cache.CacheAtomicityMode; +import org.apache.ignite.cache.CacheMode; +import org.apache.ignite.cache.query.ContinuousQueryWithTransformer; +import org.apache.ignite.cache.query.QueryCursor; +import org.apache.ignite.configuration.CacheConfiguration; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.lang.IgniteClosure; +import org.apache.ignite.testframework.GridTestUtils; +import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; +import org.junit.Test; + +/** + */ +public class CacheContinuousWithTransformerUnregisteredClassTest extends GridCommonAbstractTest { + + /** */ + private static final int DFLT_SERVER_NODE_CNT = 2; + + /** {@inheritDoc} */ + @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(igniteInstanceName); + + CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); + + ccfg.setCacheMode(CacheMode.PARTITIONED); + ccfg.setAtomicityMode(atomicityMode()); + ccfg.setBackups(1); + + cfg.setCacheConfiguration(ccfg); + + return cfg; + } + + /** + * @return Cache atomicity mode. + */ + protected CacheAtomicityMode atomicityMode() { + return CacheAtomicityMode.ATOMIC; + } + + /** {@inheritDoc} */ + @Override protected void afterTest() throws Exception { + grid(0).cache(DEFAULT_CACHE_NAME).removeAll(); + } + + /** {@inheritDoc} */ + @Override protected void beforeTestsStarted() throws Exception { + startGrids(DFLT_SERVER_NODE_CNT); + + super.beforeTestsStarted(); + } + + /** {@inheritDoc} */ + @Override protected void afterTestsStopped() throws Exception { + stopAllGrids(true); + } + + /** */ + @Test + public void testUnregisteredResultClassOnAtomicCache() throws Exception { + List received = Collections.synchronizedList(new ArrayList<>()); + + ContinuousQueryWithTransformer qry = new ContinuousQueryWithTransformer<>(); + + qry.setRemoteTransformerFactory(FactoryBuilder.factoryOf(new ToPersonId())); + + qry.setLocalListener(evts -> { + for (Object e : evts) + received.add(e); + }); + + IgniteCache cache = grid(0).cache(DEFAULT_CACHE_NAME); + + int rmtKey = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + try (QueryCursor ignored = cache.query(qry)) { + cache.put(rmtKey, 1); + cache.put(rmtKey, 2); + cache.put(rmtKey, 3); + + boolean allReceived = GridTestUtils.waitForCondition(() -> received.size() == 3, 5_000); + + assertTrue("Received: " + received, allReceived); + + for (Object o : received) + assertTrue("Unexpected class: " + o.getClass().getName(), o instanceof PersonId); + } + } + + /** */ + private static class PersonId { + /** */ + final int id; + + /** */ + PersonId(int id) { + this.id = id; + } + } + + /** + */ + private static class ToPersonId implements IgniteClosure, PersonId> { + /** */ + /** {@inheritDoc} */ + @Override public PersonId apply(CacheEntryEvent evt) { + return new PersonId(evt.getKey()); + } + } +} diff --git a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite6.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite6.java index 60d7e9954ddb4..6471274dfa7af 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite6.java +++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite6.java @@ -30,6 +30,7 @@ import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryRandomOperationsTest; import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousWithTransformerPartitionedSelfTest; import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousWithTransformerRandomOperationsTest; +import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousWithTransformerUnregisteredClassTest; import org.apache.ignite.internal.processors.cache.query.continuous.CacheKeepBinaryIterationNearEnabledTest; import org.apache.ignite.internal.processors.cache.query.continuous.CacheKeepBinaryIterationStoreEnabledTest; import org.apache.ignite.internal.processors.cache.query.continuous.CacheKeepBinaryIterationTest; @@ -61,6 +62,7 @@ IgniteInsertNullableDuplicatesSqlTest.class, GridCacheContinuousQueryPartitionAtomicOneNodeTest.class, CacheContinuousWithTransformerPartitionedSelfTest.class, + CacheContinuousWithTransformerUnregisteredClassTest.class, GridCacheContinuousQueryReplicatedAtomicOneNodeTest.class, ContinuousQueryMarshallerTest.class, GridCacheContinuousQueryReplicatedAtomicSelfTest.class, From d688c50356b3f0969644de7cbd4375bf5c8f3569 Mon Sep 17 00:00:00 2001 From: Reza Taroosheh Date: Sun, 1 Nov 2026 12:31:49 +0100 Subject: [PATCH 2/4] IGNITE-29227 Transform events for remote listeners of an ATOMIC cache in the async callback pool --- .../query/continuous/CacheContinuousQueryHandler.java | 7 ++++++- 1 file changed, 6 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..68daf61c2744d 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 @@ -605,7 +605,12 @@ public void keepBinary(boolean keepBinary) { // skipPrimaryCheck is set only when listen locally for replicated cache events. assert !skipPrimaryCheck || (cctx.isReplicated() && ctx.localNodeId().equals(nodeId)); - if (asyncCb) { + // A transformed event for a listener on another node is converted to a cache object, which may register + // the result class. An ATOMIC update calls this listener under the topology lock, where registration is + // not allowed, so such events are handled in the async callback pool. + boolean asyncTrans = rmtTransFactory != null && !nodeId.equals(ctx.localNodeId()) && cctx.atomic(); + + if (asyncCb || asyncTrans) { ContinuousQueryAsyncClosure clsr = new ContinuousQueryAsyncClosure( primary, evt, From 043fd728095c6b2c767e70747b35d24818430056 Mon Sep 17 00:00:00 2001 From: Reza Taroosheh Date: Sun, 1 Nov 2026 12:31:50 +0100 Subject: [PATCH 3/4] IGNITE-29227 Clean up the test --- ...nuousWithTransformerUnregisteredClassTest.java | 15 ++++----------- 1 file changed, 4 insertions(+), 11 deletions(-) diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerUnregisteredClassTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerUnregisteredClassTest.java index b346cdef337eb..e9c9efee76508 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerUnregisteredClassTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerUnregisteredClassTest.java @@ -35,6 +35,8 @@ import org.junit.Test; /** + * Checks that {@link ContinuousQueryWithTransformer} on an ATOMIC cache delivers events transformed on a remote node + * when the class of the transformer result is not registered yet. */ public class CacheContinuousWithTransformerUnregisteredClassTest extends GridCommonAbstractTest { @@ -48,7 +50,7 @@ public class CacheContinuousWithTransformerUnregisteredClassTest extends GridCom CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); ccfg.setCacheMode(CacheMode.PARTITIONED); - ccfg.setAtomicityMode(atomicityMode()); + ccfg.setAtomicityMode(CacheAtomicityMode.ATOMIC); ccfg.setBackups(1); cfg.setCacheConfiguration(ccfg); @@ -56,13 +58,6 @@ public class CacheContinuousWithTransformerUnregisteredClassTest extends GridCom return cfg; } - /** - * @return Cache atomicity mode. - */ - protected CacheAtomicityMode atomicityMode() { - return CacheAtomicityMode.ATOMIC; - } - /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { grid(0).cache(DEFAULT_CACHE_NAME).removeAll(); @@ -123,10 +118,8 @@ private static class PersonId { } } - /** - */ + /** */ private static class ToPersonId implements IgniteClosure, PersonId> { - /** */ /** {@inheritDoc} */ @Override public PersonId apply(CacheEntryEvent evt) { return new PersonId(evt.getKey()); From 78bd68e25875747312aa9eed35423e93dc15d6a1 Mon Sep 17 00:00:00 2001 From: Reza Taroosheh Date: Sun, 1 Nov 2026 19:17:12 +0100 Subject: [PATCH 4/4] IGNITE-29227 Run the transformer in the update thread and only convert and send in the async callback pool --- .../CacheContinuousQueryHandler.java | 79 +++++++++++++------ 1 file changed, 54 insertions(+), 25 deletions(-) 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 68daf61c2744d..dcbece2db7656 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 @@ -605,12 +605,7 @@ public void keepBinary(boolean keepBinary) { // skipPrimaryCheck is set only when listen locally for replicated cache events. assert !skipPrimaryCheck || (cctx.isReplicated() && ctx.localNodeId().equals(nodeId)); - // A transformed event for a listener on another node is converted to a cache object, which may register - // the result class. An ATOMIC update calls this listener under the topology lock, where registration is - // not allowed, so such events are handled in the async callback pool. - boolean asyncTrans = rmtTransFactory != null && !nodeId.equals(ctx.localNodeId()) && cctx.atomic(); - - if (asyncCb || asyncTrans) { + if (asyncCb) { ContinuousQueryAsyncClosure clsr = new ContinuousQueryAsyncClosure( primary, evt, @@ -1201,21 +1196,18 @@ else if (!entry.isFiltered()) notifyLocalListener(F.asList(evt), trans); } else { - if (!entry.isFiltered()) { - if (trans != null) - entry = transformToEntry(trans, evt); - - prepareEntry(cctx, nodeId, entry); - } - - Object entryOrList = handleEntry(cctx, entry); - - if (entryOrList != null) { - if (log.isDebugEnabled()) - log.debug("Send the following event to listener: " + entryOrList); - - ctx.continuous().addNotification(nodeId, routineId, entryOrList, topic, sync, true); + Object transVal = trans != null && !entry.isFiltered() ? transform(trans, evt) : null; + + // The transformer result is converted to a cache object, which may register its class. An ATOMIC update + // calls this listener under the topology lock, where registration is not allowed, so the conversion and + // sending are done in the async callback pool, ordered by partition. + if (trans != null && cctx.atomic() && !asyncCb) { + ctx.pools().asyncCallbackPool().execute( + () -> notifyRemoteListener(cctx, evt, true, transVal), + evt.partitionId()); } + else + notifyRemoteListener(cctx, evt, trans != null, transVal); } } catch (ClusterTopologyCheckedException ex) { @@ -1264,6 +1256,45 @@ else if (!entry.isFiltered()) } } + /** + * Converts the event for the listener on another node and sends it. + * + * @param cctx Cache context. + * @param evt Event. + * @param transformed Whether the query has a transformer. + * @param transVal Transformer result. + */ + private void notifyRemoteListener(GridCacheContext cctx, CacheContinuousQueryEvent evt, + boolean transformed, @Nullable Object transVal) { + try { + CacheContinuousQueryEntry entry = evt.entry(); + + if (!entry.isFiltered()) { + if (transformed) + entry = transformToEntry(transVal, evt); + + prepareEntry(cctx, nodeId, entry); + } + + Object entryOrList = handleEntry(cctx, entry); + + if (entryOrList != null) { + if (log.isDebugEnabled()) + log.debug("Send the following event to listener: " + entryOrList); + + ctx.continuous().addNotification(nodeId, routineId, entryOrList, topic, sync, true); + } + } + catch (ClusterTopologyCheckedException ex) { + if (log.isDebugEnabled()) + log.debug("Failed to send event notification to node, node left cluster " + + "[node=" + nodeId + ", err=" + ex + ']'); + } + catch (IgniteCheckedException ex) { + U.error(ctx.log(CU.CONTINUOUS_QRY_LOG_CATEGORY), "Failed to send event notification to node: " + nodeId, ex); + } + } + /** * Notifies local listener. * @@ -1712,17 +1743,15 @@ private Iterable transform(final IgniteClosure, ?> trans, + private CacheContinuousQueryEntry transformToEntry(Object transVal, CacheContinuousQueryEvent evt) { - Object transVal = transform(trans, evt); - CacheObject cacheObj = transVal == null ? null : cacheContext(ctx).toCacheObject(transVal); EventType type = evt.entry().eventType();