From 1ec3952184350b25fc40c4c9a18b8199b0397a92 Mon Sep 17 00:00:00 2001 From: Reza Taroosheh Date: Wed, 28 Oct 2026 22:24:51 +0100 Subject: [PATCH 1/3] IGNITE-29226 Add a test for transformed events on a keep-binary cache --- ...ntinuousWithTransformerKeepBinaryTest.java | 135 ++++++++++++++++++ .../IgniteCacheQuerySelfTestSuite3.java | 5 +- 2 files changed, 139 insertions(+), 1 deletion(-) create mode 100644 modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerKeepBinaryTest.java diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerKeepBinaryTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerKeepBinaryTest.java new file mode 100644 index 0000000000000..1d4c4339d2a8e --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerKeepBinaryTest.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 CacheContinuousWithTransformerKeepBinaryTest 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()); + + cfg.setCacheConfiguration(ccfg); + + return cfg; + } + + /** + * @return Cache atomicity mode. + */ + protected CacheAtomicityMode atomicityMode() { + return CacheAtomicityMode.TRANSACTIONAL; + } + + /** {@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 testKeepBinaryLocalAndRemoteEvents() 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); + + // Primary on node 0, where the query runs: its event is transformed locally. + int locKey = primaryKey(cache); + + // Primary on node 1: its event is transformed there and sent to node 0. + int rmtKey = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + try (QueryCursor ignored = cache.withKeepBinary().query(qry)) { + cache.put(locKey, locKey); + cache.put(rmtKey, rmtKey); + + assertTrue(GridTestUtils.waitForCondition(() -> received.size() == 2, 5_000)); + + 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/IgniteCacheQuerySelfTestSuite3.java b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite3.java index cb682564f263a..0028bd859bc30 100644 --- a/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite3.java +++ b/modules/indexing/src/test/java/org/apache/ignite/testsuites/IgniteCacheQuerySelfTestSuite3.java @@ -26,6 +26,7 @@ import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryExecuteInPrimaryTest; import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousQueryFactoryAsyncFilterRandomOperationTest; import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousWithTransformerClientSelfTest; +import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousWithTransformerKeepBinaryTest; import org.apache.ignite.internal.processors.cache.query.continuous.CacheContinuousWithTransformerReplicatedSelfTest; import org.apache.ignite.internal.processors.cache.query.continuous.ClientReconnectContinuousQueryTest; import org.apache.ignite.internal.processors.cache.query.continuous.ContinuousQueryReassignmentTest; @@ -73,7 +74,9 @@ CacheContinuousQueryCounterPartitionedAtomicTest.class, CacheContinuousQueryCounterPartitionedTxTest.class, CacheContinuousQueryCounterReplicatedAtomicTest.class, - CacheContinuousQueryCounterReplicatedTxTest.class + CacheContinuousQueryCounterReplicatedTxTest.class, + + CacheContinuousWithTransformerKeepBinaryTest.class }) public class IgniteCacheQuerySelfTestSuite3 { } From d9317d53a4657b6bf7d13719cda542bc8da34208 Mon Sep 17 00:00:00 2001 From: Reza Taroosheh Date: Sat, 31 Oct 2026 17:44:42 +0100 Subject: [PATCH 2/3] IGNITE-29226 Keep transformed continuous query events binary only if the transformer returns a BinaryObject --- .../continuous/CacheContinuousQueryEntry.java | 10 ++ .../CacheContinuousQueryHandler.java | 11 +- ...ntinuousWithTransformerKeepBinaryTest.java | 100 +++++++++++++++++- 3 files changed, 117 insertions(+), 4 deletions(-) diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java index 717d9431d8ee9..4bab10c355af2 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousQueryEntry.java @@ -303,6 +303,16 @@ CacheObject oldValue() { return oldVal; } + /** + * Clears the keep binary flag. + * + * @param flags Flags. + * @return Flags without the keep binary flag. + */ + static byte removeKeepBinary(byte flags) { + return (byte)(flags & ~KEEP_BINARY); + } + /** {@inheritDoc} */ @Override public void deploy(GridDeploymentInfo depInfo) { this.depInfo = depInfo; 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..b1395813bdfcd 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 @@ -40,6 +40,7 @@ import org.apache.ignite.IgniteLogger; import org.apache.ignite.IgniteSystemProperties; import org.apache.ignite.SystemProperty; +import org.apache.ignite.binary.BinaryObject; import org.apache.ignite.cache.CacheEntryEventSerializableFilter; import org.apache.ignite.cache.query.ContinuousQueryWithTransformer.EventListener; import org.apache.ignite.cluster.ClusterNode; @@ -1734,16 +1735,22 @@ private CacheContinuousQueryEntry transformToEntry(IgniteClosure received = Collections.synchronizedList(new ArrayList<>()); ContinuousQueryWithTransformer qry = new ContinuousQueryWithTransformer<>(); @@ -112,9 +114,83 @@ public void testKeepBinaryLocalAndRemoteEvents() throws Exception { } } + /** */ + @Test + public void testBinaryValueResultArrivesAsIs() throws Exception { + List received = Collections.synchronizedList(new ArrayList<>()); + + ContinuousQueryWithTransformer qry = new ContinuousQueryWithTransformer<>(); + + qry.setRemoteTransformerFactory(FactoryBuilder.factoryOf(new ToValue())); + + qry.setLocalListener(evts -> { + for (Object e : evts) + received.add(e); + }); + + IgniteCache cache = grid(0).cache(DEFAULT_CACHE_NAME); + + // Primary on node 0, where the query runs: its event is transformed locally. + int locKey = primaryKey(cache); + + // Primary on node 1: its event is transformed there and sent to node 0. + int rmtKey = primaryKey(grid(1).cache(DEFAULT_CACHE_NAME)); + + try (QueryCursor ignored = cache.withKeepBinary().query(qry)) { + cache.put(locKey, new PersonId(locKey)); + cache.put(rmtKey, new PersonId(rmtKey)); + + assertTrue(GridTestUtils.waitForCondition(() -> received.size() == 2, 5_000)); + + for (Object o : received) + assertTrue("Unexpected class: " + o.getClass().getName(), o instanceof BinaryObject); + } + } + + /** + * + */ + @Test + public void testBinaryKeyResultArrivesAsIs() throws Exception { + List received = Collections.synchronizedList(new ArrayList<>()); + + ContinuousQueryWithTransformer qry = new ContinuousQueryWithTransformer<>(); + + qry.setRemoteTransformerFactory(FactoryBuilder.factoryOf(new ToKey())); + + qry.setLocalListener(evts -> { + for (Object e : evts) + received.add(e); + }); + + IgniteCache cache = grid(0).cache(DEFAULT_CACHE_NAME); + + // Primary on node 0, where the query runs: its event is transformed locally. + PersonId locKey = new PersonId(primaryKey(cache)); + + // Primary on node 1: its event is transformed there and sent to node 0. + PersonId rmtKey = new PersonId(primaryKey(grid(1).cache(DEFAULT_CACHE_NAME))); + + // Make sure one event is transformed locally and the other remotely. + // Check that @AffinityKeyMapped keeps each key on the node primaryKey() found it for. + assertTrue(grid(0).affinity(DEFAULT_CACHE_NAME).isPrimary(grid(0).localNode(), locKey)); + assertTrue(grid(0).affinity(DEFAULT_CACHE_NAME).isPrimary(grid(1).localNode(), rmtKey)); + + try (QueryCursor ignored = cache.withKeepBinary().query(qry)) { + cache.put(locKey, locKey.id); + cache.put(rmtKey, rmtKey.id); + + assertTrue(GridTestUtils.waitForCondition(() -> received.size() == 2, 5_000)); + + for (Object o : received) + assertTrue("Unexpected class: " + o.getClass().getName(), o instanceof BinaryObject); + } + } + /** */ private static class PersonId { - /** */ + /** So the key has the same owner as its id. */ + @AffinityKeyMapped final int id; /** */ @@ -132,4 +208,24 @@ private static class ToPersonId implements IgniteClosure, Object> { + /** */ + /** {@inheritDoc} */ + @Override public Object apply(CacheEntryEvent evt) { + return evt.getValue(); + } + } + + /** + */ + private static class ToKey implements IgniteClosure, Object> { + /** */ + /** {@inheritDoc} */ + @Override public Object apply(CacheEntryEvent evt) { + return evt.getKey(); + } + } } From 322f2d73c81769f4438071ef107f176169e414b9 Mon Sep 17 00:00:00 2001 From: Reza Taroosheh Date: Sat, 31 Oct 2026 18:05:30 +0100 Subject: [PATCH 3/3] IGNITE-29226 Clean up the test --- ...ntinuousWithTransformerKeepBinaryTest.java | 27 +++++-------------- 1 file changed, 7 insertions(+), 20 deletions(-) diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerKeepBinaryTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerKeepBinaryTest.java index c8e5067ae66e0..12ae3fdea9c2d 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerKeepBinaryTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerKeepBinaryTest.java @@ -37,6 +37,8 @@ import org.junit.Test; /** + * Checks that {@link ContinuousQueryWithTransformer} on a keep-binary cache delivers the transformer result + * to the listener as it was returned, for both local and remote events. */ public class CacheContinuousWithTransformerKeepBinaryTest extends GridCommonAbstractTest { @@ -50,20 +52,13 @@ public class CacheContinuousWithTransformerKeepBinaryTest extends GridCommonAbst CacheConfiguration ccfg = new CacheConfiguration(DEFAULT_CACHE_NAME); ccfg.setCacheMode(CacheMode.PARTITIONED); - ccfg.setAtomicityMode(atomicityMode()); + ccfg.setAtomicityMode(CacheAtomicityMode.TRANSACTIONAL); cfg.setCacheConfiguration(ccfg); return cfg; } - /** - * @return Cache atomicity mode. - */ - protected CacheAtomicityMode atomicityMode() { - return CacheAtomicityMode.TRANSACTIONAL; - } - /** {@inheritDoc} */ @Override protected void afterTest() throws Exception { grid(0).cache(DEFAULT_CACHE_NAME).removeAll(); @@ -147,9 +142,7 @@ public void testBinaryValueResultArrivesAsIs() throws Exception { } } - /** - * - */ + /** */ @Test public void testBinaryKeyResultArrivesAsIs() throws Exception { List received = Collections.synchronizedList(new ArrayList<>()); @@ -199,30 +192,24 @@ private static class PersonId { } } - /** - */ + /** */ private static class ToPersonId implements IgniteClosure, PersonId> { - /** */ /** {@inheritDoc} */ @Override public PersonId apply(CacheEntryEvent evt) { return new PersonId(evt.getKey()); } } - /** - */ + /** */ private static class ToValue implements IgniteClosure, Object> { - /** */ /** {@inheritDoc} */ @Override public Object apply(CacheEntryEvent evt) { return evt.getValue(); } } - /** - */ + /** */ private static class ToKey implements IgniteClosure, Object> { - /** */ /** {@inheritDoc} */ @Override public Object apply(CacheEntryEvent evt) { return evt.getKey();