The Broker (ProducerStateManager) has to keep in memory the list of all the Aborted transactions, in order to report to Consumers the ids and let them skip the records on the client side
Currently there is no mechanism that removes such list from memory
Modifications:
Add a periodic job that for every topic that has at least one Aborted Transaction finds the first possible offset and then discards the transactions that are useless (lastOffset < offset)
In order to get the first available offset we have to read the first available entry of the topic (that may be on offloaded storage if we are using offloaders) and then get the "offset" from that
Motivation:
Modifications: