rezataroosheh commented on code in PR #13769:
URL: https://github.com/apache/ignite/pull/13769#discussion_r4454424226
##########
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) {
Review Comment:
You're not missing anything; this is deliberate.
On an ATOMIC cache, this code runs under the topology lock, so the
conversion can't register a new class. `toCacheObject` throws
`UnregisteredClassException`; the update retries with a new counter, and the
event for the old counter is lost. The listener then doesn't get the later
events from that partition.
The transformer has to stay in the update thread.
`P2PClassLoadingFailureHandlingTest` expects the put to fail if the
transformer's class can't be loaded, but the put doesn't wait for the pool, so
an error there can't fail it. My first version ran the transformer in the pool,
and this test failed on TC. Using the `asyncCb` path would move the transformer
to the pool as well, so I didn't use it.
The pool is striped by partition, so the events stay in order. The downside
is that all transformed events on ATOMIC caches now go through the pool, even
if the class is already registered.
If you see a better place for the conversion, I'm happy to change it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
