jolshan commented on a change in pull request #9590:
URL: https://github.com/apache/kafka/pull/9590#discussion_r583216706
##########
File path: core/src/main/scala/kafka/log/LogCleaner.scala
##########
@@ -701,11 +719,17 @@ private[log] class Cleaner(val id: Int,
// if any messages are to be retained, write them out
val outputBuffer = result.outputBuffer
if (outputBuffer.position() > 0) {
+ if (destSegment.isEmpty) {
+ // create a new segment with a suffix appended to the name of the
log and indexes
+ destSegment = Some(LogCleaner.createNewCleanedSegment(log,
result.minOffset()))
+ transactionMetadata.cleanedIndex = Some(destSegment.get.txnIndex)
Review comment:
I can do it in `onControlBatchRead` directly I believe. I'll add to
toAppend when there is no `cleanedIndex` and as soon as there is, I'll add
those transactions + any new ones read. From then on, I can just add the
transactions in `onControlBatchRead` as before.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
