rezataroosheh commented on code in PR #13769:
URL: https://github.com/apache/ignite/pull/13769#discussion_r4456332602
##########
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:
I looked at the other pools in `PoolProcessor`, and none seem like a better
fit. This is my understanding of the system and the issue, so please correct me
if I'm wrong. The pool has to meet two conditions:
- It has to be striped to keep the events of a partition in order.
- Its threads have to be able to wait, because the conversion may wait for a
class registration.
The striped pools are:
- `getStripedExecutorService()`: it also processes cache messages, so
waiting for a registration there would hold up other cache operations on that
stripe.
- `getDataStreamerExecutorService()` and
`getStripedRebalanceExecutorService()`: they belong to other subsystems, so
using them would be even more unexpected.
- `asyncCallbackPool()`: only continuous queries use it, so it is the
closest match.
The other option is a new pool only for this conversion, but that adds more
threads to each node and another pool size to configure.
That's why I used asyncCallbackPool(). If you prefer a separate pool, I can
do that. What do you think?
--
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]
