zstan commented on code in PR #13769:
URL: https://github.com/apache/ignite/pull/13769#discussion_r4479470682
##########
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(
Review Comment:
Seems you can use a pool only if exception is raized : 99.99% are going into
old path also cctx.atomic() && !asyncCb will become redundant:
smth like:
try {
entry = transformToEntry(transVal, evt);
}
catch (UnregisteredClassException | UnregisteredBinaryTypeException ex) {
ctx.pools().asyncCallbackPool().execute(() ->
notifyRemoteListener(cctx, evt, transVal), evt.partitionId());
}
--
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]
