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..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 @@ -1196,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) { @@ -1259,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. * @@ -1707,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(); 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..e9c9efee76508 --- /dev/null +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/query/continuous/CacheContinuousWithTransformerUnregisteredClassTest.java @@ -0,0 +1,128 @@ +/* + * 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; + +/** + * 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 { + + /** */ + 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(CacheAtomicityMode.ATOMIC); + ccfg.setBackups(1); + + cfg.setCacheConfiguration(ccfg); + + return cfg; + } + + /** {@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,