Address 2nd batch of reviews

apache · May 15, 2024 · bc27574 · bc27574
1 parent 1282e3b
commit bc27574
Show file tree

Hide file tree

Showing 11 changed files with 26 additions and 21 deletions.
diff --git a/clients/src/test/java/org/apache/kafka/common/compress/GzipCompressionTest.java b/clients/src/test/java/org/apache/kafka/common/compress/GzipCompressionTest.java
@@ -27,6 +27,7 @@
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.Collections;
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -37,13 +38,13 @@ public class GzipCompressionTest {
     @Test
     public void testCompressionDecompression() throws IOException {
         GzipCompression.Builder builder = Compression.gzip();
-        byte[] data = "data".getBytes(StandardCharsets.UTF_8);
+        byte[] data = String.join("", Collections.nCopies(256, "data")).getBytes(StandardCharsets.UTF_8);
 
         for (byte magic : Arrays.asList(RecordBatch.MAGIC_VALUE_V0, RecordBatch.MAGIC_VALUE_V1, RecordBatch.MAGIC_VALUE_V2)) {
             for (int level : Arrays.asList(GzipCompression.MIN_LEVEL, GzipCompression.DEFAULT_LEVEL, GzipCompression.MAX_LEVEL)) {
                 GzipCompression compression = builder.level(level).build();
                 ByteBufferOutputStream bufferStream = new ByteBufferOutputStream(4);
-                try (OutputStream out = compression.wrapForOutput(bufferStream, RecordBatch.CURRENT_MAGIC_VALUE)) {
+                try (OutputStream out = compression.wrapForOutput(bufferStream, magic)) {
                     out.write(data);
                     out.flush();
                 }
@@ -77,7 +78,6 @@ public void testLevelValidator() {
             validator.ensureValid("", level);
         }
         validator.ensureValid("", GzipCompression.DEFAULT_LEVEL);
-        assertThrows(ConfigException.class, () -> validator.ensureValid("", 0));
         assertThrows(ConfigException.class, () -> validator.ensureValid("", GzipCompression.MIN_LEVEL - 1));
         assertThrows(ConfigException.class, () -> validator.ensureValid("", GzipCompression.MAX_LEVEL + 1));
     }

diff --git a/clients/src/test/java/org/apache/kafka/common/compress/Lz4CompressionTest.java b/clients/src/test/java/org/apache/kafka/common/compress/Lz4CompressionTest.java
@@ -38,6 +38,7 @@
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Random;
 import java.util.stream.Stream;
@@ -85,13 +86,13 @@ public void testLz4FramingMagicV1() {
     @Test
     public void testCompressionDecompression() throws IOException {
         Lz4Compression.Builder builder = Compression.lz4();
-        byte[] data = "data".getBytes(StandardCharsets.UTF_8);
+        byte[] data = String.join("", Collections.nCopies(256, "data")).getBytes(StandardCharsets.UTF_8);
 
         for (byte magic : Arrays.asList(RecordBatch.MAGIC_VALUE_V0, RecordBatch.MAGIC_VALUE_V1, RecordBatch.MAGIC_VALUE_V2)) {
             for (int level : Arrays.asList(Lz4Compression.MIN_LEVEL, Lz4Compression.DEFAULT_LEVEL, Lz4Compression.MAX_LEVEL)) {
                 Lz4Compression compression = builder.level(level).build();
                 ByteBufferOutputStream bufferStream = new ByteBufferOutputStream(4);
-                try (OutputStream out = compression.wrapForOutput(bufferStream, RecordBatch.CURRENT_MAGIC_VALUE)) {
+                try (OutputStream out = compression.wrapForOutput(bufferStream, magic)) {
                     out.write(data);
                     out.flush();
                 }

diff --git a/clients/src/test/java/org/apache/kafka/common/compress/NoCompressionTest.java b/clients/src/test/java/org/apache/kafka/common/compress/NoCompressionTest.java
@@ -26,6 +26,7 @@
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.Collections;
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -35,11 +36,11 @@ public class NoCompressionTest {
     @Test
     public void testCompressionDecompression() throws IOException {
         NoCompression compression = Compression.NONE;
-        byte[] data = "data".getBytes(StandardCharsets.UTF_8);
+        byte[] data = String.join("", Collections.nCopies(256, "data")).getBytes(StandardCharsets.UTF_8);
 
         for (byte magic : Arrays.asList(RecordBatch.MAGIC_VALUE_V0, RecordBatch.MAGIC_VALUE_V1, RecordBatch.MAGIC_VALUE_V2)) {
             ByteBufferOutputStream bufferStream = new ByteBufferOutputStream(4);
-            try (OutputStream out = compression.wrapForOutput(bufferStream, RecordBatch.CURRENT_MAGIC_VALUE)) {
+            try (OutputStream out = compression.wrapForOutput(bufferStream, magic)) {
                 out.write(data);
                 out.flush();
             }

diff --git a/clients/src/test/java/org/apache/kafka/common/compress/SnappyCompressionTest.java b/clients/src/test/java/org/apache/kafka/common/compress/SnappyCompressionTest.java
@@ -26,6 +26,7 @@
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.Collections;
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -35,11 +36,11 @@ public class SnappyCompressionTest {
     @Test
     public void testCompressionDecompression() throws IOException {
         SnappyCompression compression = Compression.snappy().build();
-        byte[] data = "data".getBytes(StandardCharsets.UTF_8);
+        byte[] data = String.join("", Collections.nCopies(256, "data")).getBytes(StandardCharsets.UTF_8);
 
         for (byte magic : Arrays.asList(RecordBatch.MAGIC_VALUE_V0, RecordBatch.MAGIC_VALUE_V1, RecordBatch.MAGIC_VALUE_V2)) {
             ByteBufferOutputStream bufferStream = new ByteBufferOutputStream(4);
-            try (OutputStream out = compression.wrapForOutput(bufferStream, RecordBatch.CURRENT_MAGIC_VALUE)) {
+            try (OutputStream out = compression.wrapForOutput(bufferStream, magic)) {
                 out.write(data);
                 out.flush();
             }

diff --git a/clients/src/test/java/org/apache/kafka/common/compress/ZstdCompressionTest.java b/clients/src/test/java/org/apache/kafka/common/compress/ZstdCompressionTest.java
@@ -26,6 +26,7 @@
 import java.io.OutputStream;
 import java.nio.charset.StandardCharsets;
 import java.util.Arrays;
+import java.util.Collections;
 
 import static org.junit.jupiter.api.Assertions.assertArrayEquals;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -36,13 +37,13 @@ public class ZstdCompressionTest {
     @Test
     public void testCompressionDecompression() throws IOException {
         ZstdCompression.Builder builder = Compression.zstd();
-        byte[] data = "data".getBytes(StandardCharsets.UTF_8);
+        byte[] data = String.join("", Collections.nCopies(256, "data")).getBytes(StandardCharsets.UTF_8);
 
         for (byte magic : Arrays.asList(RecordBatch.MAGIC_VALUE_V0, RecordBatch.MAGIC_VALUE_V1, RecordBatch.MAGIC_VALUE_V2)) {
             for (int level : Arrays.asList(ZstdCompression.MIN_LEVEL, ZstdCompression.DEFAULT_LEVEL, ZstdCompression.MAX_LEVEL)) {
                 ZstdCompression compression = builder.level(level).build();
                 ByteBufferOutputStream bufferStream = new ByteBufferOutputStream(4);
-                try (OutputStream out = compression.wrapForOutput(bufferStream, RecordBatch.CURRENT_MAGIC_VALUE)) {
+                try (OutputStream out = compression.wrapForOutput(bufferStream, magic)) {
                     out.write(data);
                     out.flush();
                 }

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
@@ -74,7 +74,7 @@ public Args(int bufferOffset, Compression compression, byte magic) {
         public String toString() {
             return "magic=" + magic +
                 ", bufferOffset=" + bufferOffset +
-                ", compressionType=" + compression;
+                ", compression=" + compression;
         }
     }
 

diff --git a/core/src/main/scala/kafka/log/UnifiedLog.scala b/core/src/main/scala/kafka/log/UnifiedLog.scala
@@ -37,6 +37,7 @@ import org.apache.kafka.server.common.{MetadataVersion, OffsetAndEpoch}
 import org.apache.kafka.server.common.MetadataVersion.IBP_0_10_0_IV0
 import org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig
 import org.apache.kafka.server.metrics.KafkaMetricsGroup
+import org.apache.kafka.server.record.BrokerCompressionType
 import org.apache.kafka.server.util.Scheduler
 import org.apache.kafka.storage.internals.checkpoint.{LeaderEpochCheckpointFile, PartitionMetadataFile}
 import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache
@@ -790,7 +791,7 @@ class UnifiedLog(@volatile var logStartOffset: Long,
             val offset = PrimitiveRef.ofLong(localLog.logEndOffset)
             appendInfo.setFirstOffset(offset.value)
             val validateAndOffsetAssignResult = try {
-              val targetCompression = config.compressionType.targetCompression(config.compression, appendInfo.sourceCompression())
+              val targetCompression = BrokerCompressionType.targetCompression(config.compression, appendInfo.sourceCompression())
               val validator = new LogValidator(validRecords,
                 topicPartition,
                 time,

diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
@@ -96,6 +96,7 @@ class LogConfigTest {
       case TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG => assertPropertyInvalid(name, "not_a_number", "-3")
       case TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG => assertPropertyInvalid(name, "not_a_number", "-3")
       case TopicConfig.COMPRESSION_GZIP_LEVEL_CONFIG => assertPropertyInvalid(name, "not_a_number", "-2")
+      case TopicConfig.COMPRESSION_LZ4_LEVEL_CONFIG => assertPropertyInvalid(name, "not_a_number", "-1")
       case TopicConfig.COMPRESSION_ZSTD_LEVEL_CONFIG => assertPropertyInvalid(name, "not_a_number", "-0.1")
 
       case _ => assertPropertyInvalid(name, "not_a_number", "-1")

diff --git a/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala b/core/src/test/scala/unit/kafka/log/LogValidatorTest.scala
@@ -96,12 +96,12 @@ class LogValidatorTest {
 
   private def checkOnlyOneBatch(magic: Byte, sourceCompression: Compression, targetCompression: Compression): Unit = {
     assertThrows(classOf[InvalidRecordException],
-      () => validateMessages(createTwoBatchedRecords(magic, 0L, sourceCompression), magic, sourceCompression.`type`(), targetCompression)
+      () => validateMessages(createTwoBatchedRecords(magic, sourceCompression), magic, sourceCompression.`type`(), targetCompression)
     )
   }
 
   private def checkAllowMultiBatch(magic: Byte, sourceCompression: Compression, targetCompression: Compression): Unit = {
-    validateMessages(createTwoBatchedRecords(magic, 0L, sourceCompression), magic, sourceCompression.`type`(), targetCompression)
+    validateMessages(createTwoBatchedRecords(magic, sourceCompression), magic, sourceCompression.`type`(), targetCompression)
   }
 
   private def checkMismatchMagic(batchMagic: Byte, recordMagic: Byte, compression: Compression): Unit = {
@@ -1610,9 +1610,7 @@ class LogValidatorTest {
     builder.build()
   }
 
-  private def createTwoBatchedRecords(magicValue: Byte,
-                                      timestamp: Long,
-                                      codec: Compression): MemoryRecords = {
+  private def createTwoBatchedRecords(magicValue: Byte, codec: Compression): MemoryRecords = {
     val buf = ByteBuffer.allocate(2048)
     var builder = MemoryRecords.builder(buf, magicValue, codec, TimestampType.CREATE_TIME, 0L)
     builder.append(10L, "1".getBytes(), "a".getBytes())

diff --git a/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala b/core/src/test/scala/unit/kafka/server/KafkaConfigTest.scala
@@ -714,7 +714,7 @@ class KafkaConfigTest {
   def testInvalidGzipCompressionLevel(): Unit = {
     val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
     props.setProperty(KafkaConfig.CompressionTypeProp, "gzip")
-    props.setProperty(KafkaConfig.CompressionGzipLevelProp, (GzipCompression.MAX_LEVEL+1).toString)
+    props.setProperty(KafkaConfig.CompressionGzipLevelProp, (GzipCompression.MAX_LEVEL + 1).toString)
     assertThrows(classOf[ConfigException], () => KafkaConfig.fromProps(props))
   }
 
@@ -942,7 +942,8 @@ class KafkaConfigTest {
         case MetricConfigs.METRIC_RECORDING_LEVEL_CONFIG => // ignore string
 
         case KafkaConfig.CompressionGzipLevelProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "0")
-        case KafkaConfig.CompressionZstdLevelProp => assertPropertyInvalid(baseProperties, name, "not_a_number", ZstdCompression.MAX_LEVEL+1)
+        case KafkaConfig.CompressionLz4LevelProp => assertPropertyInvalid(baseProperties, name, "not_a_number", "0")
+        case KafkaConfig.CompressionZstdLevelProp => assertPropertyInvalid(baseProperties, name, "not_a_number", ZstdCompression.MAX_LEVEL + 1)
 
         case KafkaConfig.RackProp => // ignore string
         //SSL Configs

diff --git a/server-common/src/main/java/org/apache/kafka/server/record/BrokerCompressionType.java b/server-common/src/main/java/org/apache/kafka/server/record/BrokerCompressionType.java
@@ -42,7 +42,7 @@ public enum BrokerCompressionType {
         this.name = name;
     }
 
-    public Compression targetCompression(Optional<Compression> logCompression, CompressionType producerCompressionType) {
+    public static Compression targetCompression(Optional<Compression> logCompression, CompressionType producerCompressionType) {
         return logCompression.orElseGet(() -> Compression.of(producerCompressionType).build());
     }