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<>(); + + 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); + } + } + + /** */ + @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; + + /** */ + 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()); + } + } + + /** */ + 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(); + } + } +} 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 { }