MINOR: Use streaming iterator with decompression buffer when building offset map
This makes it consistent with the `filterTo` methods.
@@ -938,7 +938,9 @@ private[log] class Cleaner(val id: Int,
// Note that abort markers are supported in v2 and above, which means count is defined.
stats.indexMessagesRead(batch.countOrNull)
} else {
- for (record <- batch.asScala) {
+ val recordsIterator = batch.streamingIterator(decompressionBufferSupplier)
+ try {
+ for (record <- recordsIterator.asScala) {
if (record.hasKey && record.offset >= startOffset) {
if (map.size < maxDesiredMapSize)
map.put(record.key, record.offset)
@@ -947,6 +949,7 @@ private[log] class Cleaner(val id: Int,
}
stats.indexMessagesRead(1)
}
+ } finally recordsIterator.close()
}
}
