aliehsaeedii commented on PR #16554:
URL: https://github.com/apache/kafka/pull/16554#issuecomment-3841544479
@abhi-ksolves Thanks for follwing up.
As you mentioned what ends up CME is in fact direct manipulation of the
`sortedMap` which could be removing or putting. The question was where does
that happen? In the former rounds, I missed the `callback`. It seems that
`callback` is the culprit. This is the scenario:
1. evictWhile is iterating over the sortedMap
2. It calls the `emit` callback (see `evictwhile` call in
`KTableSuppressProcessorSupplier`)
4. `emit` calls `context.forward()` which pushes the record downstream
5. Downstream processing can trigger additional operations such as remvoing
from or putting to sortedMap
6. When control returns to the evictWhile, the iterator detects the
modification and throws ConcurrentModificationException
This is particularly problematic during state restoration (as mentioned in
the ticket) because, restoration processes many records in batches.
So the solution must put the `callback` outside of the iteration loop.
--
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]
