rezataroosheh commented on code in PR #13769:
URL: https://github.com/apache/ignite/pull/13769#discussion_r4477822445
##########
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());
Review Comment:
I checked this, and it isn't a problem here.
`CacheContinuousQueryManager.onEntryUpdated()` creates a new entry and a new
event for each listener and each update, and the update thread doesn't change
them after this call. `ContinuousQueryAsyncClosure` already passes the same
event to this pool in the same way.
--
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]
