This is an automated email from the ASF dual-hosted git repository. showuon pushed a commit to branch 3.7 in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/3.7 by this push: new 1ee20e3aa0f KAFKA-16342 fix getOffsetByMaxTimestamp for compressed records (#15474) 1ee20e3aa0f is described below commit 1ee20e3aa0fbe541ed7629388ce84e4a7b9c33b3 Author: Luke Chen <show...@gmail.com> AuthorDate: Fri Mar 15 06:09:45 2024 +0800 KAFKA-16342 fix getOffsetByMaxTimestamp for compressed records (#15474) Fix getOffsetByMaxTimestamp for compressed records. This PR adds: 1) For inPlaceAssignment case, compute the correct offset for maxTimestamp when traversing the batch records, and set to ValidationResult in the end, instead of setting to last offset always. 2) For not inPlaceAssignment, set the offsetOfMaxTimestamp for the log create time, like non-compressed, and inPlaceAssignment cases, instead of setting to last offset always. 3) Add tests to verify the fix. Reviewers: Jun Rao <jun...@apache.org>, Chia-Ping Tsai <chia7...@gmail.com> --- .../kafka/common/record/MemoryRecordsBuilder.java | 29 +-- .../common/record/MemoryRecordsBuilderTest.java | 23 ++- .../kafka/common/record/MemoryRecordsTest.java | 5 +- .../kafka/admin/ListOffsetsIntegrationTest.scala | 202 ++++++++++++++++++--- .../kafka/server/QuorumTestHarness.scala | 2 +- .../scala/unit/kafka/log/LogValidatorTest.scala | 29 +-- .../kafka/storage/internals/log/LogValidator.java | 13 +- .../org/apache/kafka/tools/GetOffsetShellTest.java | 11 +- 8 files changed, 233 insertions(+), 81 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java index 054d72ee822..88991328bb7 100644 --- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java +++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecordsBuilder.java @@ -242,34 +242,23 @@ public class MemoryRecordsBuilder implements AutoCloseable { /** * Get the max timestamp and its offset. The details of the offset returned are a bit subtle. + * Note: The semantic for the offset of max timestamp is the first offset with the max timestamp if there are multi-records having same timestamp. * - * If the log append time is used, the offset will be the last offset unless no compression is used and - * the message format version is 0 or 1, in which case, it will be the first offset. + * If the log append time is used, the offset will be the first offset of the record. * - * If create time is used, the offset will be the last offset unless no compression is used and the message - * format version is 0 or 1, in which case, it will be the offset of the record with the max timestamp. + * If create time is used, the offset will always be the offset of the record with the max timestamp. + * + * If it's NO_TIMESTAMP (i.e. MAGIC_VALUE_V0), we'll return offset -1 since no timestamp info in records. * * @return The max timestamp and its offset */ public RecordsInfo info() { if (timestampType == TimestampType.LOG_APPEND_TIME) { - long shallowOffsetOfMaxTimestamp; - // Use the last offset when dealing with record batches - if (compressionType != CompressionType.NONE || magic >= RecordBatch.MAGIC_VALUE_V2) - shallowOffsetOfMaxTimestamp = lastOffset; - else - shallowOffsetOfMaxTimestamp = baseOffset; - return new RecordsInfo(logAppendTime, shallowOffsetOfMaxTimestamp); - } else if (maxTimestamp == RecordBatch.NO_TIMESTAMP) { - return new RecordsInfo(RecordBatch.NO_TIMESTAMP, lastOffset); + return new RecordsInfo(logAppendTime, baseOffset); } else { - long shallowOffsetOfMaxTimestamp; - // Use the last offset when dealing with record batches - if (compressionType != CompressionType.NONE || magic >= RecordBatch.MAGIC_VALUE_V2) - shallowOffsetOfMaxTimestamp = lastOffset; - else - shallowOffsetOfMaxTimestamp = offsetOfMaxTimestamp; - return new RecordsInfo(maxTimestamp, shallowOffsetOfMaxTimestamp); + // For create time, we always use offsetOfMaxTimestamp for the correct time -> offset mapping + // If it's MAGIC_VALUE_V0, the value will be the default value: [-1, -1] + return new RecordsInfo(maxTimestamp, offsetOfMaxTimestamp); } } diff --git a/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsBuilderTest.java b/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsBuilderTest.java index 5616fb23f7d..eb28ba47ef5 100644 --- a/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsBuilderTest.java +++ b/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsBuilderTest.java @@ -377,11 +377,8 @@ public class MemoryRecordsBuilderTest { MemoryRecordsBuilder.RecordsInfo info = builder.info(); assertEquals(logAppendTime, info.maxTimestamp); - - if (args.compressionType == CompressionType.NONE && magic <= MAGIC_VALUE_V1) - assertEquals(0L, info.shallowOffsetOfMaxTimestamp); - else - assertEquals(2L, info.shallowOffsetOfMaxTimestamp); + // When logAppendTime is used, the first offset of the batch will be the offset of maxTimestamp + assertEquals(0L, info.shallowOffsetOfMaxTimestamp); for (RecordBatch batch : records.batches()) { if (magic == MAGIC_VALUE_V0) { @@ -415,10 +412,11 @@ public class MemoryRecordsBuilderTest { assertEquals(2L, info.maxTimestamp); } - if (args.compressionType == CompressionType.NONE && magic == MAGIC_VALUE_V1) - assertEquals(1L, info.shallowOffsetOfMaxTimestamp); + if (magic == MAGIC_VALUE_V0) + // in MAGIC_VALUE_V0's case, we don't have timestamp info in records, so always return -1. + assertEquals(-1L, info.shallowOffsetOfMaxTimestamp); else - assertEquals(2L, info.shallowOffsetOfMaxTimestamp); + assertEquals(1L, info.shallowOffsetOfMaxTimestamp); int i = 0; long[] expectedTimestamps = new long[] {0L, 2L, 1L}; @@ -495,12 +493,13 @@ public class MemoryRecordsBuilderTest { MemoryRecords records = builder.build(); MemoryRecordsBuilder.RecordsInfo info = builder.info(); - if (magic == MAGIC_VALUE_V0) + if (magic == MAGIC_VALUE_V0) { assertEquals(-1, info.maxTimestamp); - else + assertEquals(-1L, info.shallowOffsetOfMaxTimestamp); + } else { assertEquals(2L, info.maxTimestamp); - - assertEquals(2L, info.shallowOffsetOfMaxTimestamp); + assertEquals(2L, info.shallowOffsetOfMaxTimestamp); + } long i = 0L; for (RecordBatch batch : records.batches()) { diff --git a/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java b/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java index 3f0195bf5d1..50821af841c 100644 --- a/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java +++ b/clients/src/test/java/org/apache/kafka/common/record/MemoryRecordsTest.java @@ -893,10 +893,7 @@ public class MemoryRecordsTest { assertEquals(filtered.limit(), result.bytesRetained()); if (magic > RecordBatch.MAGIC_VALUE_V0) { assertEquals(20L, result.maxTimestamp()); - if (compression == CompressionType.NONE && magic < RecordBatch.MAGIC_VALUE_V2) - assertEquals(4L, result.shallowOffsetOfMaxTimestamp()); - else - assertEquals(5L, result.shallowOffsetOfMaxTimestamp()); + assertEquals(4L, result.shallowOffsetOfMaxTimestamp()); } MemoryRecords filteredRecords = MemoryRecords.readableRecords(filtered); diff --git a/core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala b/core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala index 333bd980a60..e5e22e9dff9 100644 --- a/core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala +++ b/core/src/test/scala/integration/kafka/admin/ListOffsetsIntegrationTest.scala @@ -19,82 +19,236 @@ package kafka.admin import kafka.integration.KafkaServerTestHarness import kafka.server.KafkaConfig +import kafka.utils.TestUtils.{createProducer, plaintextBootstrapServers} import kafka.utils.{TestInfoUtils, TestUtils} import org.apache.kafka.clients.admin._ import org.apache.kafka.clients.producer.ProducerRecord import org.apache.kafka.common.TopicPartition -import org.apache.kafka.common.utils.Utils +import org.apache.kafka.common.config.TopicConfig +import org.apache.kafka.common.utils.{MockTime, Time, Utils} import org.junit.jupiter.api.Assertions.assertEquals import org.junit.jupiter.api.{AfterEach, BeforeEach, TestInfo} import org.junit.jupiter.params.ParameterizedTest import org.junit.jupiter.params.provider.ValueSource +import java.util.Properties import scala.collection.{Map, Seq} import scala.jdk.CollectionConverters._ class ListOffsetsIntegrationTest extends KafkaServerTestHarness { val topicName = "foo" + val topicNameWithCustomConfigs = "foo2" var adminClient: Admin = _ + var setOldMessageFormat: Boolean = false + val mockTime: Time = new MockTime(1) @BeforeEach override def setUp(testInfo: TestInfo): Unit = { super.setUp(testInfo) - createTopic(topicName, 1, 1.toShort) - produceMessages() + createTopicWithConfig(topicName, new Properties()) adminClient = Admin.create(Map[String, Object]( AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG -> bootstrapServers() ).asJava) } + override def brokerTime(brokerId: Int): Time = mockTime + @AfterEach override def tearDown(): Unit = { + setOldMessageFormat = false Utils.closeQuietly(adminClient, "ListOffsetsAdminClient") super.tearDown() } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @ValueSource(strings = Array("zk", "kraft")) - def testEarliestOffset(quorum: String): Unit = { - val earliestOffset = runFetchOffsets(adminClient, OffsetSpec.earliest()) - assertEquals(0, earliestOffset.offset()) + def testThreeCompressedRecordsInOneBatch(quorum: String): Unit = { + produceMessagesInOneBatch("gzip") + verifyListOffsets() + + // test LogAppendTime case + val props: Properties = new Properties() + props.setProperty(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "LogAppendTime") + createTopicWithConfig(topicNameWithCustomConfigs, props) + produceMessagesInOneBatch("gzip", topicNameWithCustomConfigs) + // In LogAppendTime's case, the maxTimestampOffset should be the first message of the batch. + // So in this one batch test, it'll be the first offset 0 + verifyListOffsets(topic = topicNameWithCustomConfigs, 0) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @ValueSource(strings = Array("zk", "kraft")) - def testLatestOffset(quorum: String): Unit = { - val latestOffset = runFetchOffsets(adminClient, OffsetSpec.latest()) - assertEquals(3, latestOffset.offset()) + def testThreeRecordsInSeparateBatch(quorum: String): Unit = { + produceMessagesInSeparateBatch() + verifyListOffsets() + } + + // The message conversion test only run in ZK mode because KRaft mode doesn't support "inter.broker.protocol.version" < 3.0 + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) + @ValueSource(strings = Array("zk")) + def testThreeRecordsInOneBatchWithMessageConversion(quorum: String): Unit = { + createOldMessageFormatBrokers() + produceMessagesInOneBatch() + verifyListOffsets() + + // test LogAppendTime case + val props: Properties = new Properties() + props.setProperty(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "LogAppendTime") + createTopicWithConfig(topicNameWithCustomConfigs, props) + produceMessagesInOneBatch(topic = topicNameWithCustomConfigs) + // In LogAppendTime's case, the maxTimestampOffset should be the first message of the batch. + // So in this one batch test, it'll be the first offset 0 + verifyListOffsets(topic = topicNameWithCustomConfigs, 0) + } + + // The message conversion test only run in ZK mode because KRaft mode doesn't support "inter.broker.protocol.version" < 3.0 + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) + @ValueSource(strings = Array("zk")) + def testThreeRecordsInSeparateBatchWithMessageConversion(quorum: String): Unit = { + createOldMessageFormatBrokers() + produceMessagesInSeparateBatch() + verifyListOffsets() + + // test LogAppendTime case + val props: Properties = new Properties() + props.setProperty(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "LogAppendTime") + createTopicWithConfig(topicNameWithCustomConfigs, props) + produceMessagesInSeparateBatch(topic = topicNameWithCustomConfigs) + // In LogAppendTime's case, the maxTimestampOffset should be the first message of the batch. + // So in this separate batch test, it'll be the last offset 2 + verifyListOffsets(topic = topicNameWithCustomConfigs, 2) + } + + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) + @ValueSource(strings = Array("zk", "kraft")) + def testThreeRecordsInOneBatchHavingDifferentCompressionTypeWithServer(quorum: String): Unit = { + val props: Properties = new Properties() + props.setProperty(TopicConfig.COMPRESSION_TYPE_CONFIG, "lz4") + createTopicWithConfig(topicNameWithCustomConfigs, props) + produceMessagesInOneBatch(topic = topicNameWithCustomConfigs) + verifyListOffsets(topic = topicNameWithCustomConfigs) + } + + @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) + @ValueSource(strings = Array("zk", "kraft")) + def testThreeRecordsInSeparateBatchHavingDifferentCompressionTypeWithServer(quorum: String): Unit = { + val props: Properties = new Properties() + props.setProperty(TopicConfig.COMPRESSION_TYPE_CONFIG, "lz4") + createTopicWithConfig(topicNameWithCustomConfigs, props) + produceMessagesInSeparateBatch(topic = topicNameWithCustomConfigs) + verifyListOffsets(topic = topicNameWithCustomConfigs) } @ParameterizedTest(name = TestInfoUtils.TestWithParameterizedQuorumName) @ValueSource(strings = Array("zk", "kraft")) - def testMaxTimestampOffset(quorum: String): Unit = { - val maxTimestampOffset = runFetchOffsets(adminClient, OffsetSpec.maxTimestamp()) - assertEquals(1, maxTimestampOffset.offset()) + def testThreeCompressedRecordsInSeparateBatch(quorum: String): Unit = { + produceMessagesInSeparateBatch("gzip") + verifyListOffsets() + + // test LogAppendTime case + val props: Properties = new Properties() + props.setProperty(TopicConfig.MESSAGE_TIMESTAMP_TYPE_CONFIG, "LogAppendTime") + createTopicWithConfig(topicNameWithCustomConfigs, props) + produceMessagesInSeparateBatch("gzip", topicNameWithCustomConfigs) + // In LogAppendTime's case, the maxTimestampOffset should be the first message of the batch. + // So in this separate batch test, it'll be the last offset 2 + verifyListOffsets(topic = topicNameWithCustomConfigs, 2) + } + + private def createOldMessageFormatBrokers(): Unit = { + setOldMessageFormat = true + recreateBrokers(reconfigure = true, startup = true) + Utils.closeQuietly(adminClient, "ListOffsetsAdminClient") + adminClient = Admin.create(Map[String, Object]( + AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG -> bootstrapServers() + ).asJava) + } + + private def createTopicWithConfig(topic: String, props: Properties): Unit = { + createTopic(topic, 1, 1.toShort, topicConfig = props) + } + + private def verifyListOffsets(topic: String = topicName, expectedMaxTimestampOffset: Int = 1): Unit = { + val earliestOffset = runFetchOffsets(adminClient, OffsetSpec.earliest(), topic) + assertEquals(0, earliestOffset.offset()) + + val latestOffset = runFetchOffsets(adminClient, OffsetSpec.latest(), topic) + assertEquals(3, latestOffset.offset()) + + val maxTimestampOffset = runFetchOffsets(adminClient, OffsetSpec.maxTimestamp(), topic) + assertEquals(expectedMaxTimestampOffset, maxTimestampOffset.offset()) } private def runFetchOffsets(adminClient: Admin, - offsetSpec: OffsetSpec): ListOffsetsResult.ListOffsetsResultInfo = { - val tp = new TopicPartition(topicName, 0) + offsetSpec: OffsetSpec, + topic: String): ListOffsetsResult.ListOffsetsResultInfo = { + val tp = new TopicPartition(topic, 0) adminClient.listOffsets(Map( tp -> offsetSpec ).asJava, new ListOffsetsOptions()).all().get().get(tp) } - def produceMessages(): Unit = { + private def produceMessagesInOneBatch(compressionType: String = "none", topic: String = topicName): Unit = { val records = Seq( - new ProducerRecord[Array[Byte], Array[Byte]](topicName, 0, 100L, - null, new Array[Byte](10000)), - new ProducerRecord[Array[Byte], Array[Byte]](topicName, 0, 999L, - null, new Array[Byte](10000)), - new ProducerRecord[Array[Byte], Array[Byte]](topicName, 0, 200L, - null, new Array[Byte](10000)), + new ProducerRecord[Array[Byte], Array[Byte]](topic, 0, 100L, + null, new Array[Byte](10)), + new ProducerRecord[Array[Byte], Array[Byte]](topic, 0, 999L, + null, new Array[Byte](10)), + new ProducerRecord[Array[Byte], Array[Byte]](topic, 0, 200L, + null, new Array[Byte](10)), ) - TestUtils.produceMessages(brokers, records, -1) + // create a producer with large linger.ms and enough batch.size (default is enough for three 10 bytes records), + // so that we can confirm all records will be accumulated in producer until we flush them into one batch. + val producer = createProducer( + plaintextBootstrapServers(brokers), + deliveryTimeoutMs = Int.MaxValue, + lingerMs = Int.MaxValue, + compressionType = compressionType) + + try { + val futures = records.map(producer.send) + producer.flush() + futures.foreach(_.get) + } finally { + producer.close() + } } - def generateConfigs: Seq[KafkaConfig] = - TestUtils.createBrokerConfigs(1, zkConnectOrNull).map(KafkaConfig.fromProps) + private def produceMessagesInSeparateBatch(compressionType: String = "none", topic: String = topicName): Unit = { + val records = Seq(new ProducerRecord[Array[Byte], Array[Byte]](topic, 0, 100L, + null, new Array[Byte](10))) + val records2 = Seq(new ProducerRecord[Array[Byte], Array[Byte]](topic, 0, 999L, + null, new Array[Byte](10))) + val records3 = Seq(new ProducerRecord[Array[Byte], Array[Byte]](topic, 0, 200L, + null, new Array[Byte](10))) + + val producer = createProducer( + plaintextBootstrapServers(brokers), + compressionType = compressionType) + try { + val futures = records.map(producer.send) + futures.foreach(_.get) + // advance the server time after each record sent to make sure the time changed when appendTime is used + mockTime.sleep(100) + val futures2 = records2.map(producer.send) + futures2.foreach(_.get) + mockTime.sleep(100) + val futures3 = records3.map(producer.send) + futures3.foreach(_.get) + } finally { + producer.close() + } + } + + def generateConfigs: Seq[KafkaConfig] = { + TestUtils.createBrokerConfigs(1, zkConnectOrNull).map{ props => + if (setOldMessageFormat) { + props.setProperty("log.message.format.version", "0.10.0") + props.setProperty("inter.broker.protocol.version", "0.10.0") + } + props + }.map(KafkaConfig.fromProps) + } } diff --git a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala index 1a83809c0ad..4d2ff87483f 100755 --- a/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala +++ b/core/src/test/scala/integration/kafka/server/QuorumTestHarness.scala @@ -124,7 +124,7 @@ class KRaftQuorumImplementation( util.EnumSet.of(REQUIRE_AT_LEAST_ONE_VALID, REQUIRE_METADATA_LOG_DIR)) val sharedServer = new SharedServer(config, metaPropertiesEnsemble, - Time.SYSTEM, + time, new Metrics(), controllerQuorumVotersFuture, faultHandlerFactory) diff --git a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala index 0b5d17cbcc2..ac6152b9b15 100644 --- a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala @@ -219,8 +219,8 @@ class LogValidatorTest { "MessageSet should still valid") assertEquals(now, validatedResults.maxTimestampMs, s"Max timestamp should be $now") - assertEquals(records.records.asScala.size - 1, validatedResults.shallowOffsetOfMaxTimestampMs, - s"The offset of max timestamp should be ${records.records.asScala.size - 1}") + assertEquals(0, validatedResults.shallowOffsetOfMaxTimestampMs, + s"The offset of max timestamp should be 0 if logAppendTime is used") assertTrue(validatedResults.messageSizeMaybeChanged, "Message size may have been changed") @@ -271,8 +271,8 @@ class LogValidatorTest { "MessageSet should still valid") assertEquals(now, validatedResults.maxTimestampMs, s"Max timestamp should be $now") - assertEquals(records.records.asScala.size - 1, validatedResults.shallowOffsetOfMaxTimestampMs, - s"The offset of max timestamp should be ${records.records.asScala.size - 1}") + assertEquals(0, validatedResults.shallowOffsetOfMaxTimestampMs, + s"The offset of max timestamp should be 0 if logAppendTime is used") assertFalse(validatedResults.messageSizeMaybeChanged, "Message size should not have been changed") @@ -341,6 +341,7 @@ class LogValidatorTest { private def checkNonCompressed(magic: Byte): Unit = { val now = System.currentTimeMillis() + // set the timestamp of seq(1) (i.e. offset 1) as the max timestamp val timestampSeq = Seq(now - 1, now + 1, now) val (producerId, producerEpoch, baseSequence, isTransactional, partitionLeaderEpoch) = @@ -431,6 +432,7 @@ class LogValidatorTest { private def checkRecompression(magic: Byte): Unit = { val now = System.currentTimeMillis() + // set the timestamp of seq(1) (i.e. offset 1) as the max timestamp val timestampSeq = Seq(now - 1, now + 1, now) val (producerId, producerEpoch, baseSequence, isTransactional, partitionLeaderEpoch) = @@ -484,8 +486,8 @@ class LogValidatorTest { } assertEquals(now + 1, validatingResults.maxTimestampMs, s"Max timestamp should be ${now + 1}") - assertEquals(2, validatingResults.shallowOffsetOfMaxTimestampMs, - "Offset of max timestamp should be 2") + assertEquals(1, validatingResults.shallowOffsetOfMaxTimestampMs, + "Offset of max timestamp should be 1") assertTrue(validatingResults.messageSizeMaybeChanged, "Message size should have been changed") @@ -536,8 +538,8 @@ class LogValidatorTest { } assertEquals(validatedResults.maxTimestampMs, RecordBatch.NO_TIMESTAMP, s"Max timestamp should be ${RecordBatch.NO_TIMESTAMP}") - assertEquals(validatedRecords.records.asScala.size - 1, validatedResults.shallowOffsetOfMaxTimestampMs, - s"Offset of max timestamp should be ${validatedRecords.records.asScala.size - 1}") + assertEquals(-1, validatedResults.shallowOffsetOfMaxTimestampMs, + s"Offset of max timestamp should be -1") assertTrue(validatedResults.messageSizeMaybeChanged, "Message size should have been changed") verifyRecordValidationStats(validatedResults.recordValidationStats, numConvertedRecords = 3, records, @@ -583,8 +585,8 @@ class LogValidatorTest { assertEquals(RecordBatch.NO_SEQUENCE, batch.baseSequence) } assertEquals(timestamp, validatedResults.maxTimestampMs) - assertEquals(validatedRecords.records.asScala.size - 1, validatedResults.shallowOffsetOfMaxTimestampMs, - s"Offset of max timestamp should be ${validatedRecords.records.asScala.size - 1}") + assertEquals(0, validatedResults.shallowOffsetOfMaxTimestampMs, + s"Offset of max timestamp should be 0 when multiple records having the same max timestamp.") assertTrue(validatedResults.messageSizeMaybeChanged, "Message size should have been changed") verifyRecordValidationStats(validatedResults.recordValidationStats, numConvertedRecords = 3, records, @@ -598,6 +600,7 @@ class LogValidatorTest { private def checkCompressed(magic: Byte): Unit = { val now = System.currentTimeMillis() + // set the timestamp of seq(1) (i.e. offset 1) as the max timestamp val timestampSeq = Seq(now - 1, now + 1, now) val (producerId, producerEpoch, baseSequence, isTransactional, partitionLeaderEpoch) = @@ -654,11 +657,9 @@ class LogValidatorTest { } assertEquals(now + 1, validatedResults.maxTimestampMs, s"Max timestamp should be ${now + 1}") - // All versions have an outer batch when compressed, so the shallow offset - // of max timestamp is always the offset of the last record in the batch. - val expectedShallowOffsetOfMaxTimestamp = recordList.size - 1 + val expectedShallowOffsetOfMaxTimestamp = 1 assertEquals(expectedShallowOffsetOfMaxTimestamp, validatedResults.shallowOffsetOfMaxTimestampMs, - s"Offset of max timestamp should be ${validatedRecords.records.asScala.size - 1}") + s"Offset of max timestamp should be 1") assertFalse(validatedResults.messageSizeMaybeChanged, "Message size should not have been changed") verifyRecordValidationStats(validatedResults.recordValidationStats, numConvertedRecords = 0, records, diff --git a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java index 378247fa270..0cf9cd1c60f 100644 --- a/storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java +++ b/storage/src/main/java/org/apache/kafka/storage/internals/log/LogValidator.java @@ -330,6 +330,8 @@ public class LogValidator { long maxTimestamp = RecordBatch.NO_TIMESTAMP; LongRef expectedInnerOffset = PrimitiveRef.ofLong(0); List<Record> validatedRecords = new ArrayList<>(); + long offsetOfMaxTimestamp = -1; + long initialOffset = offsetCounter.value; int uncompressedSizeInBytes = 0; @@ -379,8 +381,11 @@ public class LogValidator { && batch.magic() > RecordBatch.MAGIC_VALUE_V0 && toMagic > RecordBatch.MAGIC_VALUE_V0) { - if (record.timestamp() > maxTimestamp) + if (record.timestamp() > maxTimestamp) { maxTimestamp = record.timestamp(); + // The offset is only increased when it is a valid record + offsetOfMaxTimestamp = initialOffset + validatedRecords.size(); + } // Some older clients do not implement the V1 internal offsets correctly. // Historically the broker handled this by rewriting the batches rather @@ -417,8 +422,10 @@ public class LogValidator { long lastOffset = offsetCounter.value - 1; firstBatch.setLastOffset(lastOffset); - if (timestampType == TimestampType.LOG_APPEND_TIME) + if (timestampType == TimestampType.LOG_APPEND_TIME) { maxTimestamp = now; + offsetOfMaxTimestamp = initialOffset; + } if (toMagic >= RecordBatch.MAGIC_VALUE_V1) firstBatch.setMaxTimestamp(timestampType, maxTimestamp); @@ -431,7 +438,7 @@ public class LogValidator { now, records, maxTimestamp, - lastOffset, + offsetOfMaxTimestamp, false, recordValidationStats); } diff --git a/tools/src/test/java/org/apache/kafka/tools/GetOffsetShellTest.java b/tools/src/test/java/org/apache/kafka/tools/GetOffsetShellTest.java index 417bbe71168..ce9c1718ffb 100644 --- a/tools/src/test/java/org/apache/kafka/tools/GetOffsetShellTest.java +++ b/tools/src/test/java/org/apache/kafka/tools/GetOffsetShellTest.java @@ -30,6 +30,7 @@ import org.apache.kafka.clients.producer.ProducerConfig; import org.apache.kafka.clients.producer.ProducerRecord; import org.apache.kafka.common.serialization.StringSerializer; import org.apache.kafka.common.utils.Exit; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Tag; import org.junit.jupiter.api.extension.ExtendWith; @@ -52,20 +53,24 @@ public class GetOffsetShellTest { private final int topicCount = 4; private final int offsetTopicPartitionCount = 4; private final ClusterInstance cluster; + private final String topicName = "topic"; public GetOffsetShellTest(ClusterInstance cluster) { this.cluster = cluster; } private String getTopicName(int i) { - return "topic" + i; + return topicName + i; } - public void setUp() { + @BeforeEach + public void before() { cluster.config().serverProperties().put("auto.create.topics.enable", false); cluster.config().serverProperties().put("offsets.topic.replication.factor", "1"); cluster.config().serverProperties().put("offsets.topic.num.partitions", String.valueOf(offsetTopicPartitionCount)); + } + private void setUp() { try (Admin admin = Admin.create(cluster.config().adminClientProperties())) { List<NewTopic> topics = new ArrayList<>(); @@ -333,7 +338,7 @@ public class GetOffsetShellTest { } private List<Row> expectedOffsetsWithInternal() { - List<Row> consOffsets = IntStream.range(0, offsetTopicPartitionCount + 1) + List<Row> consOffsets = IntStream.range(0, offsetTopicPartitionCount) .mapToObj(i -> new Row("__consumer_offsets", i, 0L)) .collect(Collectors.toList());