From d38251ffe9a16644439108d14566be0cd1554466 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Fri, 24 May 2019 22:13:13 -0700 Subject: [PATCH 1/7] always assume a single batch with compressed messages --- .../main/scala/kafka/log/LogValidator.scala | 148 +++++++++--------- 1 file changed, 78 insertions(+), 70 deletions(-) diff --git a/core/src/main/scala/kafka/log/LogValidator.scala b/core/src/main/scala/kafka/log/LogValidator.scala index 8bf215ef79844..310cbf311cf56 100644 --- a/core/src/main/scala/kafka/log/LogValidator.scala +++ b/core/src/main/scala/kafka/log/LogValidator.scala @@ -249,83 +249,91 @@ private[kafka] object LogValidator extends Logging { partitionLeaderEpoch: Int, isFromClient: Boolean, interBrokerProtocolVersion: ApiVersion): ValidationAndOffsetAssignResult = { - // No in place assignment situation 1 and 2 - var inPlaceAssignment = sourceCodec == targetCodec && toMagic > RecordBatch.MAGIC_VALUE_V0 - - var maxTimestamp = RecordBatch.NO_TIMESTAMP - val expectedInnerOffset = new LongRef(0) - val validatedRecords = new mutable.ArrayBuffer[Record] - - var uncompressedSizeInBytes = 0 - - for (batch <- records.batches.asScala) { - validateBatch(batch, isFromClient, toMagic) - uncompressedSizeInBytes += AbstractRecords.recordBatchHeaderSizeInBytes(toMagic, batch.compressionType()) - - // Do not compress control records unless they are written compressed - if (sourceCodec == NoCompressionCodec && batch.isControlBatch) - inPlaceAssignment = true - - for (record <- batch.asScala) { - if (sourceCodec != NoCompressionCodec && record.isCompressed) - throw new InvalidRecordException("Compressed outer record should not have an inner record with a " + - s"compression attribute set: $record") - if (targetCodec == ZStdCompressionCodec && interBrokerProtocolVersion < KAFKA_2_1_IV0) - throw new UnsupportedCompressionTypeException("Produce requests to inter.broker.protocol.version < 2.1 broker " + "are not allowed to use ZStandard compression") - validateRecord(batch, record, now, timestampType, timestampDiffMaxMs, compactedTopic) - - uncompressedSizeInBytes += record.sizeInBytes() - if (batch.magic > RecordBatch.MAGIC_VALUE_V0 && toMagic > RecordBatch.MAGIC_VALUE_V0) { - // Check if we need to overwrite offset - // No in place assignment situation 3 - if (record.offset != expectedInnerOffset.getAndIncrement()) - inPlaceAssignment = false - if (record.timestamp > maxTimestamp) - maxTimestamp = record.timestamp - } - - // No in place assignment situation 4 - if (!record.hasMagic(toMagic)) - inPlaceAssignment = false - - validatedRecords += record - } - } - if (!inPlaceAssignment) { - val (producerId, producerEpoch, sequence, isTransactional) = { - // note that we only reassign offsets for requests coming straight from a producer. For records with magic V2, - // there should be exactly one RecordBatch per request, so the following is all we need to do. For Records - // with older magic versions, there will never be a producer id, etc. - val first = records.batches.asScala.head - (first.producerId, first.producerEpoch, first.baseSequence, first.isTransactional) - } - buildRecordsAndAssignOffsets(toMagic, offsetCounter, time, timestampType, CompressionType.forId(targetCodec.codec), now, - validatedRecords, producerId, producerEpoch, sequence, isTransactional, partitionLeaderEpoch, isFromClient, - uncompressedSizeInBytes) - } else { - // we can update the batch only and write the compressed payload as is - val batch = records.batches.iterator.next() - val lastOffset = offsetCounter.addAndGet(validatedRecords.size) - 1 + // No in place assignment situation 1 and 2 + var inPlaceAssignment = sourceCodec == targetCodec && toMagic > RecordBatch.MAGIC_VALUE_V0 - batch.setLastOffset(lastOffset) + var maxTimestamp = RecordBatch.NO_TIMESTAMP + val expectedInnerOffset = new LongRef(0) + val validatedRecords = new mutable.ArrayBuffer[Record] - if (timestampType == TimestampType.LOG_APPEND_TIME) - maxTimestamp = now + var uncompressedSizeInBytes = 0 - if (toMagic >= RecordBatch.MAGIC_VALUE_V1) - batch.setMaxTimestamp(timestampType, maxTimestamp) + // Assume there's only one batch with compressed memory records; otherwise, return InvalidRecordException + val batchIterator = records.batches.iterator + val batch = batchIterator.next() - if (toMagic >= RecordBatch.MAGIC_VALUE_V2) - batch.setPartitionLeaderEpoch(partitionLeaderEpoch) + if (batchIterator.hasNext) { + throw new InvalidRecordException("Compressed outer record should only have a single record batch") + } - val recordConversionStats = new RecordConversionStats(uncompressedSizeInBytes, 0, 0) - ValidationAndOffsetAssignResult(validatedRecords = records, - maxTimestamp = maxTimestamp, - shallowOffsetOfMaxTimestamp = lastOffset, - messageSizeMaybeChanged = false, - recordConversionStats = recordConversionStats) + validateBatch(batch, isFromClient, toMagic) + uncompressedSizeInBytes += AbstractRecords.recordBatchHeaderSizeInBytes(toMagic, batch.compressionType()) + + // Do not compress control records unless they are written compressed + if (sourceCodec == NoCompressionCodec && batch.isControlBatch) + inPlaceAssignment = true + + for (record <- batch.asScala) { + if (sourceCodec != NoCompressionCodec && record.isCompressed) + throw new InvalidRecordException("Compressed outer record should not have an inner record with a " + + s"compression attribute set: $record") + if (targetCodec == ZStdCompressionCodec && interBrokerProtocolVersion < KAFKA_2_1_IV0) + throw new UnsupportedCompressionTypeException("Produce requests to inter.broker.protocol.version < 2.1 broker " + "are not allowed to use ZStandard compression") + validateRecord(batch, record, now, timestampType, timestampDiffMaxMs, compactedTopic) + + uncompressedSizeInBytes += record.sizeInBytes() + if (batch.magic > RecordBatch.MAGIC_VALUE_V0 && toMagic > RecordBatch.MAGIC_VALUE_V0) { + // Check if we need to overwrite offset + // No in place assignment situation 3 + if (record.offset != expectedInnerOffset.getAndIncrement()) + inPlaceAssignment = false + if (record.timestamp > maxTimestamp) + maxTimestamp = record.timestamp } + + // No in place assignment situation 4 + if (!record.hasMagic(toMagic)) + inPlaceAssignment = false + + validatedRecords += record + } + + if (!inPlaceAssignment) { + val (producerId, producerEpoch, sequence, isTransactional) = { + // note that we only reassign offsets for requests coming straight from a producer. For records with magic V2, + // there should be exactly one RecordBatch per request, so the following is all we need to do. For Records + // with older magic versions, there will never be a producer id, etc. + val first = records.batches.asScala.head + (first.producerId, first.producerEpoch, first.baseSequence, first.isTransactional) + } + buildRecordsAndAssignOffsets(toMagic, offsetCounter, time, timestampType, CompressionType.forId(targetCodec.codec), now, + validatedRecords, producerId, producerEpoch, sequence, isTransactional, partitionLeaderEpoch, isFromClient, + uncompressedSizeInBytes) + } else { + // we can update the batch only and write the compressed payload as is; + // again we assume only one record batch within the compressed set + val batch = records.batches.iterator.next() + val lastOffset = offsetCounter.addAndGet(validatedRecords.size) - 1 + + batch.setLastOffset(lastOffset) + + if (timestampType == TimestampType.LOG_APPEND_TIME) + maxTimestamp = now + + if (toMagic >= RecordBatch.MAGIC_VALUE_V1) + batch.setMaxTimestamp(timestampType, maxTimestamp) + + if (toMagic >= RecordBatch.MAGIC_VALUE_V2) + batch.setPartitionLeaderEpoch(partitionLeaderEpoch) + + val recordConversionStats = new RecordConversionStats(uncompressedSizeInBytes, 0, 0) + ValidationAndOffsetAssignResult(validatedRecords = records, + maxTimestamp = maxTimestamp, + shallowOffsetOfMaxTimestamp = lastOffset, + messageSizeMaybeChanged = false, + recordConversionStats = recordConversionStats) + } } private def buildRecordsAndAssignOffsets(magic: Byte, From a94d2062ae11259010c2e854975c3a7a8438ce17 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 28 May 2019 12:20:59 -0700 Subject: [PATCH 2/7] add unit tests --- .../kafka/common/record/MemoryRecords.java | 18 +++++++++++++++ .../main/scala/kafka/log/LogValidator.scala | 20 +++++++++++------ .../unit/kafka/log/LogValidatorTest.scala | 22 +++++++++++++++++++ 3 files changed, 53 insertions(+), 7 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java index c6db18d419162..18ac5318ec74a 100644 --- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java +++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java @@ -628,6 +628,24 @@ public static MemoryRecords withRecords(byte magic, long initialOffset, Compress return builder.build(); } + // for testing only, create a new memory-records that contains duplicated + // bytes (therefore duplicated batches) of the passed in memory records + public static MemoryRecords duplicateRecords(MemoryRecords inputRecords) { + ByteBuffer inputBuffer = inputRecords.buffer().duplicate (); + inputBuffer.limit (inputBuffer.position() + inputBuffer.capacity()); + ByteBuffer outputBuffer = ByteBuffer.allocate(inputBuffer.capacity() * 2); + outputBuffer.put(inputBuffer); + + // write again + inputBuffer = inputRecords.buffer().duplicate (); + inputBuffer.limit (inputBuffer.position() + inputBuffer.capacity()); + outputBuffer.put(inputBuffer); + + outputBuffer.flip(); + outputBuffer.position(0); + return MemoryRecords.readableRecords(outputBuffer.slice()); + } + public static MemoryRecords withEndTransactionMarker(long producerId, short producerEpoch, EndTransactionMarker marker) { return withEndTransactionMarker(0L, System.currentTimeMillis(), RecordBatch.NO_PARTITION_LEADER_EPOCH, producerId, producerEpoch, marker); diff --git a/core/src/main/scala/kafka/log/LogValidator.scala b/core/src/main/scala/kafka/log/LogValidator.scala index 310cbf311cf56..e09645565177b 100644 --- a/core/src/main/scala/kafka/log/LogValidator.scala +++ b/core/src/main/scala/kafka/log/LogValidator.scala @@ -74,6 +74,18 @@ private[kafka] object LogValidator extends Logging { } } + private[kafka] def validateRecords(records: MemoryRecords): RecordBatch = { + // Assume there's only one batch with compressed memory records; otherwise, return InvalidRecordException + val batchIterator = records.batches.iterator + val batch = batchIterator.next() + + if (batchIterator.hasNext) { + throw new InvalidRecordException("Compressed outer record should only have a single record batch") + } + + batch + } + private def validateBatch(batch: RecordBatch, isFromClient: Boolean, toMagic: Byte): Unit = { if (isFromClient) { if (batch.magic >= RecordBatch.MAGIC_VALUE_V2) { @@ -259,13 +271,7 @@ private[kafka] object LogValidator extends Logging { var uncompressedSizeInBytes = 0 - // Assume there's only one batch with compressed memory records; otherwise, return InvalidRecordException - val batchIterator = records.batches.iterator - val batch = batchIterator.next() - - if (batchIterator.hasNext) { - throw new InvalidRecordException("Compressed outer record should only have a single record batch") - } + val batch = validateRecords(records) validateBatch(batch, isFromClient, toMagic) uncompressedSizeInBytes += AbstractRecords.recordBatchHeaderSizeInBytes(toMagic, batch.compressionType()) diff --git a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala index 37553b9376e1a..326d6562ce89a 100644 --- a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala @@ -36,6 +36,28 @@ class LogValidatorTest { val time = Time.SYSTEM + @Test(expected = classOf[InvalidRecordException]) + def testOnlyOneBatchCompressedV0(): Unit = { + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V0, CompressionType.GZIP) + } + + @Test(expected = classOf[InvalidRecordException]) + def testOnlyOneBatchCompressedV1(): Unit = { + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V1, CompressionType.SNAPPY) + } + + @Test(expected = classOf[InvalidRecordException]) + def testOnlyOneBatchCompressedV2(): Unit = { + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V2, CompressionType.LZ4) + } + + private def checkOnlyOneBatchCompressed(magic: Byte, codec: CompressionType) { + val records = createRecords(magic, 0L, codec) + val duplicates = MemoryRecords.duplicateRecords(records) + + LogValidator.validateRecords(duplicates) + } + @Test def testLogAppendTimeNonCompressedV1() { checkLogAppendTimeNonCompressed(RecordBatch.MAGIC_VALUE_V1) From 918b4d59873135ceafa7d15efebda8025874bf08 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 28 May 2019 12:46:56 -0700 Subject: [PATCH 3/7] add unit tests 2 --- .../unit/kafka/log/LogValidatorTest.scala | 32 +++++++++++-------- 1 file changed, 18 insertions(+), 14 deletions(-) diff --git a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala index 326d6562ce89a..cfac1fcbf6e3e 100644 --- a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala @@ -22,13 +22,13 @@ import java.util.concurrent.TimeUnit import kafka.api.{ApiVersion, KAFKA_2_0_IV1} import kafka.common.LongRef import kafka.message._ -import org.apache.kafka.common.errors.{InvalidTimestampException, UnsupportedCompressionTypeException, UnsupportedForMessageFormatException} +import org.apache.kafka.common.errors.{InvalidTimestampException, KafkaStorageException, UnsupportedCompressionTypeException, UnsupportedForMessageFormatException} import org.apache.kafka.common.record._ import org.apache.kafka.common.utils.Time import org.apache.kafka.test.TestUtils import org.junit.Assert._ import org.junit.Test -import org.scalatest.Assertions.intercept +import org.scalatest.Assertions.{assertThrows, intercept} import scala.collection.JavaConverters._ @@ -36,26 +36,30 @@ class LogValidatorTest { val time = Time.SYSTEM - @Test(expected = classOf[InvalidRecordException]) - def testOnlyOneBatchCompressedV0(): Unit = { - checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V0, CompressionType.GZIP) + @Test + def testOnlyOneBatchV0(): Unit = { + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V0) } - @Test(expected = classOf[InvalidRecordException]) - def testOnlyOneBatchCompressedV1(): Unit = { - checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V1, CompressionType.SNAPPY) + @Test + def testOnlyOneBatchV1(): Unit = { + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V1) } - @Test(expected = classOf[InvalidRecordException]) - def testOnlyOneBatchCompressedV2(): Unit = { - checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V2, CompressionType.LZ4) + @Test + def testOnlyOneBatchV2(): Unit = { + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V2) } - private def checkOnlyOneBatchCompressed(magic: Byte, codec: CompressionType) { - val records = createRecords(magic, 0L, codec) + private def checkOnlyOneBatchCompressed(magic: Byte) { + val records = createRecords(magic, 0L, CompressionType.GZIP) val duplicates = MemoryRecords.duplicateRecords(records) - LogValidator.validateRecords(duplicates) + LogValidator.validateRecords(records) + + assertThrows[InvalidRecordException] { + LogValidator.validateRecords(duplicates) + } } @Test From 8a6de04c63d4e72a8976238c12453ac3c06e78cd Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 28 May 2019 14:18:25 -0700 Subject: [PATCH 4/7] github comments --- .../kafka/common/record/MemoryRecords.java | 18 ----------- .../main/scala/kafka/log/LogValidator.scala | 11 +++++-- .../unit/kafka/log/LogValidatorTest.scala | 30 ++++++++++++++----- 3 files changed, 30 insertions(+), 29 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java index 18ac5318ec74a..c6db18d419162 100644 --- a/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java +++ b/clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java @@ -628,24 +628,6 @@ public static MemoryRecords withRecords(byte magic, long initialOffset, Compress return builder.build(); } - // for testing only, create a new memory-records that contains duplicated - // bytes (therefore duplicated batches) of the passed in memory records - public static MemoryRecords duplicateRecords(MemoryRecords inputRecords) { - ByteBuffer inputBuffer = inputRecords.buffer().duplicate (); - inputBuffer.limit (inputBuffer.position() + inputBuffer.capacity()); - ByteBuffer outputBuffer = ByteBuffer.allocate(inputBuffer.capacity() * 2); - outputBuffer.put(inputBuffer); - - // write again - inputBuffer = inputRecords.buffer().duplicate (); - inputBuffer.limit (inputBuffer.position() + inputBuffer.capacity()); - outputBuffer.put(inputBuffer); - - outputBuffer.flip(); - outputBuffer.position(0); - return MemoryRecords.readableRecords(outputBuffer.slice()); - } - public static MemoryRecords withEndTransactionMarker(long producerId, short producerEpoch, EndTransactionMarker marker) { return withEndTransactionMarker(0L, System.currentTimeMillis(), RecordBatch.NO_PARTITION_LEADER_EPOCH, producerId, producerEpoch, marker); diff --git a/core/src/main/scala/kafka/log/LogValidator.scala b/core/src/main/scala/kafka/log/LogValidator.scala index e09645565177b..602cb0e37e3c8 100644 --- a/core/src/main/scala/kafka/log/LogValidator.scala +++ b/core/src/main/scala/kafka/log/LogValidator.scala @@ -74,13 +74,18 @@ private[kafka] object LogValidator extends Logging { } } - private[kafka] def validateRecords(records: MemoryRecords): RecordBatch = { + private[kafka] def validateCompressedRecords(records: MemoryRecords): RecordBatch = { // Assume there's only one batch with compressed memory records; otherwise, return InvalidRecordException val batchIterator = records.batches.iterator + + if (!batchIterator.hasNext) { + throw new InvalidRecordException("Compressed outer record has no batches at all") + } + val batch = batchIterator.next() if (batchIterator.hasNext) { - throw new InvalidRecordException("Compressed outer record should only have a single record batch") + throw new InvalidRecordException("Compressed outer record has more than one batch") } batch @@ -271,7 +276,7 @@ private[kafka] object LogValidator extends Logging { var uncompressedSizeInBytes = 0 - val batch = validateRecords(records) + val batch = validateCompressedRecords(records) validateBatch(batch, isFromClient, toMagic) uncompressedSizeInBytes += AbstractRecords.recordBatchHeaderSizeInBytes(toMagic, batch.compressionType()) diff --git a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala index cfac1fcbf6e3e..57c6ecf42f2cc 100644 --- a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala @@ -37,28 +37,25 @@ class LogValidatorTest { val time = Time.SYSTEM @Test - def testOnlyOneBatchV0(): Unit = { + def testOnlyOneBatchCompressedV0(): Unit = { checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V0) } @Test - def testOnlyOneBatchV1(): Unit = { + def testOnlyOneBatchCompressedV1(): Unit = { checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V1) } @Test - def testOnlyOneBatchV2(): Unit = { + def testOnlyOneBatchCompressedV2(): Unit = { checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V2) } private def checkOnlyOneBatchCompressed(magic: Byte) { - val records = createRecords(magic, 0L, CompressionType.GZIP) - val duplicates = MemoryRecords.duplicateRecords(records) - - LogValidator.validateRecords(records) + LogValidator.validateCompressedRecords(createRecords(magic, 0L, CompressionType.GZIP)) assertThrows[InvalidRecordException] { - LogValidator.validateRecords(duplicates) + LogValidator.validateCompressedRecords(createTwoBatchedRecords(magic, 0L, CompressionType.GZIP)) } } @@ -1163,6 +1160,23 @@ class LogValidatorTest { builder.build() } + def createTwoBatchedRecords(magicValue: Byte, + timestamp: Long = RecordBatch.NO_TIMESTAMP, + codec: CompressionType): MemoryRecords = { + val buf = ByteBuffer.allocate(2048) + var builder = MemoryRecords.builder(buf, magicValue, codec, TimestampType.CREATE_TIME, 0L) + builder.append(10L, "1".getBytes(), "a".getBytes()) + builder.close() + + builder = MemoryRecords.builder(buf, magicValue, codec, TimestampType.CREATE_TIME, 1L) + builder.append(11L, "2".getBytes(), "b".getBytes()) + builder.append(12L, "3".getBytes(), "c".getBytes()) + builder.close() + + buf.flip() + MemoryRecords.readableRecords(buf.slice()) + } + /* check that offsets are assigned consecutively from the given base offset */ def checkOffsets(records: MemoryRecords, baseOffset: Long) { assertTrue("Message set should not be empty", records.records.asScala.nonEmpty) From 888733a7d63b6a6cc43b4bbcbb6bfb3d67794795 Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 28 May 2019 14:35:47 -0700 Subject: [PATCH 5/7] remove line --- core/src/test/scala/unit/kafka/log/LogValidatorTest.scala | 1 - 1 file changed, 1 deletion(-) diff --git a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala index 57c6ecf42f2cc..bece2677c607a 100644 --- a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala @@ -1167,7 +1167,6 @@ class LogValidatorTest { var builder = MemoryRecords.builder(buf, magicValue, codec, TimestampType.CREATE_TIME, 0L) builder.append(10L, "1".getBytes(), "a".getBytes()) builder.close() - builder = MemoryRecords.builder(buf, magicValue, codec, TimestampType.CREATE_TIME, 1L) builder.append(11L, "2".getBytes(), "b".getBytes()) builder.append(12L, "3".getBytes(), "c".getBytes()) From 6f05943ec52a7dac28e4382469993cf544f00a7a Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 28 May 2019 17:06:15 -0700 Subject: [PATCH 6/7] more comments --- .../src/main/scala/kafka/api/ApiVersion.scala | 2 + .../main/scala/kafka/log/LogValidator.scala | 8 +++- .../unit/kafka/log/LogValidatorTest.scala | 38 +++++++++++++++---- 3 files changed, 38 insertions(+), 10 deletions(-) diff --git a/core/src/main/scala/kafka/api/ApiVersion.scala b/core/src/main/scala/kafka/api/ApiVersion.scala index 8e8ca0b749302..ebb21744eef0e 100644 --- a/core/src/main/scala/kafka/api/ApiVersion.scala +++ b/core/src/main/scala/kafka/api/ApiVersion.scala @@ -316,6 +316,8 @@ case object KAFKA_2_3_IV1 extends DefaultApiVersion { val id: Int = 23 } + + object ApiVersionValidator extends Validator { override def ensureValid(name: String, value: Any): Unit = { diff --git a/core/src/main/scala/kafka/log/LogValidator.scala b/core/src/main/scala/kafka/log/LogValidator.scala index 602cb0e37e3c8..77a4b2bc3cf41 100644 --- a/core/src/main/scala/kafka/log/LogValidator.scala +++ b/core/src/main/scala/kafka/log/LogValidator.scala @@ -74,7 +74,7 @@ private[kafka] object LogValidator extends Logging { } } - private[kafka] def validateCompressedRecords(records: MemoryRecords): RecordBatch = { + private[kafka] def validateOneBatchRecords(records: MemoryRecords): RecordBatch = { // Assume there's only one batch with compressed memory records; otherwise, return InvalidRecordException val batchIterator = records.batches.iterator @@ -198,6 +198,10 @@ private[kafka] object LogValidator extends Logging { val initialOffset = offsetCounter.value for (batch <- records.batches.asScala) { + if (batch.magic() >= RecordBatch.MAGIC_VALUE_V2) { + validateOneBatchRecords(records) + } + validateBatch(batch, isFromClient, magic) var maxBatchTimestamp = RecordBatch.NO_TIMESTAMP @@ -276,7 +280,7 @@ private[kafka] object LogValidator extends Logging { var uncompressedSizeInBytes = 0 - val batch = validateCompressedRecords(records) + val batch = validateOneBatchRecords(records) validateBatch(batch, isFromClient, toMagic) uncompressedSizeInBytes += AbstractRecords.recordBatchHeaderSizeInBytes(toMagic, batch.compressionType()) diff --git a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala index bece2677c607a..d306b162804cb 100644 --- a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala +++ b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala @@ -19,10 +19,10 @@ package kafka.log import java.nio.ByteBuffer import java.util.concurrent.TimeUnit -import kafka.api.{ApiVersion, KAFKA_2_0_IV1} +import kafka.api.{ApiVersion, KAFKA_2_0_IV1, KAFKA_2_3_IV1} import kafka.common.LongRef import kafka.message._ -import org.apache.kafka.common.errors.{InvalidTimestampException, KafkaStorageException, UnsupportedCompressionTypeException, UnsupportedForMessageFormatException} +import org.apache.kafka.common.errors.{InvalidTimestampException, UnsupportedCompressionTypeException, UnsupportedForMessageFormatException} import org.apache.kafka.common.record._ import org.apache.kafka.common.utils.Time import org.apache.kafka.test.TestUtils @@ -38,27 +38,49 @@ class LogValidatorTest { @Test def testOnlyOneBatchCompressedV0(): Unit = { - checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V0) + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V0, CompressionType.GZIP) } @Test def testOnlyOneBatchCompressedV1(): Unit = { - checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V1) + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V1, CompressionType.GZIP) } @Test def testOnlyOneBatchCompressedV2(): Unit = { - checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V2) + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V2, CompressionType.GZIP) } - private def checkOnlyOneBatchCompressed(magic: Byte) { - LogValidator.validateCompressedRecords(createRecords(magic, 0L, CompressionType.GZIP)) + @Test + def testOnlyOneBatchUncompressedV2(): Unit = { + checkOnlyOneBatchCompressed(RecordBatch.MAGIC_VALUE_V2, CompressionType.NONE) + } + + private def checkOnlyOneBatchCompressed(magic: Byte, compressionType: CompressionType) { + validateMessages(createRecords(magic, 0L, compressionType), magic, compressionType) assertThrows[InvalidRecordException] { - LogValidator.validateCompressedRecords(createTwoBatchedRecords(magic, 0L, CompressionType.GZIP)) + validateMessages(createTwoBatchedRecords(magic, 0L, compressionType), magic, compressionType) } } + private def validateMessages(records: MemoryRecords, magic: Byte, compressionType: CompressionType): Unit = { + LogValidator.validateMessagesAndAssignOffsets(records, + new LongRef(0L), + time, + now = 0L, + CompressionCodec.getCompressionCodec(compressionType.name), + CompressionCodec.getCompressionCodec(compressionType.name), + compactedTopic = false, + magic, + TimestampType.CREATE_TIME, + 1000L, + RecordBatch.NO_PRODUCER_EPOCH, + isFromClient = true, + KAFKA_2_3_IV1 + ) + } + @Test def testLogAppendTimeNonCompressedV1() { checkLogAppendTimeNonCompressed(RecordBatch.MAGIC_VALUE_V1) From 7fa8c5522d00b97583dd5d8dd54b1391e209625a Mon Sep 17 00:00:00 2001 From: Guozhang Wang Date: Tue, 28 May 2019 21:34:48 -0700 Subject: [PATCH 7/7] remove lines --- core/src/main/scala/kafka/api/ApiVersion.scala | 2 -- 1 file changed, 2 deletions(-) diff --git a/core/src/main/scala/kafka/api/ApiVersion.scala b/core/src/main/scala/kafka/api/ApiVersion.scala index ebb21744eef0e..8e8ca0b749302 100644 --- a/core/src/main/scala/kafka/api/ApiVersion.scala +++ b/core/src/main/scala/kafka/api/ApiVersion.scala @@ -316,8 +316,6 @@ case object KAFKA_2_3_IV1 extends DefaultApiVersion { val id: Int = 23 } - - object ApiVersionValidator extends Validator { override def ensureValid(name: String, value: Any): Unit = {