ijuma commented on code in PR #18321:
URL: https://github.com/apache/kafka/pull/18321#discussion_r1904847796
##########
core/src/test/scala/unit/kafka/log/LogCleanerParameterizedIntegrationTest.scala:
##########
@@ -134,6 +135,131 @@ class LogCleanerParameterizedIntegrationTest extends
AbstractLogCleanerIntegrati
assertEquals(toMap(messages), toMap(read), "Contents of the map shouldn't
change")
}
+ @ParameterizedTest
+ @ArgumentsSource(classOf[LogCleanerParameterizedIntegrationTest.ExcludeZstd])
+ def testCleanerWithMessageFormatV0V1V2(compressionType: CompressionType):
Unit = {
+ val compression = Compression.of(compressionType).build()
+ val largeMessageKey = 20
+ val (largeMessageValue, largeMessageSet) =
createLargeSingleMessageSet(largeMessageKey, RecordBatch.MAGIC_VALUE_V0,
compression)
+ val maxMessageSize = compression match {
+ case Compression.NONE => largeMessageSet.sizeInBytes
+ case _ =>
+ // the broker assigns absolute offsets for message format 0 which
potentially causes the compressed size to
+ // increase because the broker offsets are larger than the ones
assigned by the client
+ // adding `6` to the message set size is good enough for this test: it
covers the increased message size while
+ // still being less than the overhead introduced by the conversion
from message format version 0 to 1
+ largeMessageSet.sizeInBytes + 6
Review Comment:
Will try to remove it - this was the existing code and it may have been
written before this was allowed.
--
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: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
