From 926c7b605c3ce33c2a718ebbd881a66a5fcf058b Mon Sep 17 00:00:00 2001 From: Ismael Juma Date: Mon, 6 May 2019 06:53:49 -0700 Subject: [PATCH 1/3] MINOR: Remove workarounds for lz4-java bug affecting byte buffers (#6679) lz4/lz4-java#65 was included in lz4-java 1.4.0. Relying on existing tests for verification. Reviewers: Manikumar Reddy --- .../compress/KafkaLZ4BlockInputStream.java | 33 +++---------------- 1 file changed, 4 insertions(+), 29 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java b/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java index 85e7f7b9bec9e..4fa7985b9bdcf 100644 --- a/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java +++ b/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java @@ -78,11 +78,6 @@ public KafkaLZ4BlockInputStream(ByteBuffer in, BufferSupplier bufferSupplier, bo this.bufferSupplier = bufferSupplier; readHeader(); decompressionBuffer = bufferSupplier.get(maxBlockSize); - if (!decompressionBuffer.hasArray() || decompressionBuffer.arrayOffset() != 0) { - // require array backed decompression buffer with zero offset - // to simplify workaround for https://github.com/lz4/lz4-java/pull/65 - throw new RuntimeException("decompression buffer must have backing array with zero array offset"); - } finished = false; } @@ -132,10 +127,7 @@ private void readHeader() throws IOException { int len = in.position() - in.reset().position(); - int hash = in.hasArray() ? - // workaround for https://github.com/lz4/lz4-java/pull/65 - CHECKSUM.hash(in.array(), in.arrayOffset() + in.position(), len, 0) : - CHECKSUM.hash(in, in.position(), len, 0); + int hash = CHECKSUM.hash(in, in.position(), len, 0); in.position(in.position() + len); if (in.get() != (byte) ((hash >> 8) & 0xFF)) { throw new IOException(DESCRIPTOR_HASH_MISMATCH); @@ -173,22 +165,8 @@ private void readBlock() throws IOException { if (compressed) { try { - // workaround for https://github.com/lz4/lz4-java/pull/65 - final int bufferSize; - if (in.hasArray()) { - bufferSize = DECOMPRESSOR.decompress( - in.array(), - in.position() + in.arrayOffset(), - blockSize, - decompressionBuffer.array(), - 0, - maxBlockSize - ); - } else { - // decompressionBuffer has zero arrayOffset, so we don't need to worry about - // https://github.com/lz4/lz4-java/pull/65 - bufferSize = DECOMPRESSOR.decompress(in, in.position(), blockSize, decompressionBuffer, 0, maxBlockSize); - } + final int bufferSize = DECOMPRESSOR.decompress(in, in.position(), blockSize, decompressionBuffer, 0, + maxBlockSize); decompressionBuffer.position(0); decompressionBuffer.limit(bufferSize); decompressedBuffer = decompressionBuffer; @@ -202,10 +180,7 @@ private void readBlock() throws IOException { // verify checksum if (flg.isBlockChecksumSet()) { - // workaround for https://github.com/lz4/lz4-java/pull/65 - int hash = in.hasArray() ? - CHECKSUM.hash(in.array(), in.arrayOffset() + in.position(), blockSize, 0) : - CHECKSUM.hash(in, in.position(), blockSize, 0); + int hash = CHECKSUM.hash(in, in.position(), blockSize, 0); in.position(in.position() + blockSize); if (hash != in.getInt()) { throw new IOException(BLOCK_HASH_MISMATCH); From 03b53ad984f4266e317d6e591259b2a572bf62f5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Xavier=20L=C3=A9aut=C3=A9?= Date: Tue, 23 Feb 2021 10:28:55 -0800 Subject: [PATCH 2/3] KAFKA-9203: add checks to detect buggy lz4 library versions This check allows us to safely remove the workarounds for buggy LZ4 versions without users encountering cryptic errors if they accidentally have an older LZ4 library on the classpath as described in KAFKA-9203. With this change the use will get a clear error message indicating what the problem might be if they encounter this situation. --- .../compress/KafkaLZ4BlockInputStream.java | 50 +++++++++++++++++++ 1 file changed, 50 insertions(+) diff --git a/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java b/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java index 4fa7985b9bdcf..d0ff033c764ac 100644 --- a/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java +++ b/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java @@ -16,6 +16,7 @@ */ package org.apache.kafka.common.compress; +import net.jpountz.lz4.LZ4Compressor; import net.jpountz.lz4.LZ4Exception; import net.jpountz.lz4.LZ4Factory; import net.jpountz.lz4.LZ4SafeDecompressor; @@ -51,6 +52,19 @@ public final class KafkaLZ4BlockInputStream extends InputStream { private static final LZ4SafeDecompressor DECOMPRESSOR = LZ4Factory.fastestInstance().safeDecompressor(); private static final XXHash32 CHECKSUM = XXHashFactory.fastestInstance().hash32(); + private static final RuntimeException BROKEN_LZ4_EXCEPTION; + // https://issues.apache.org/jira/browse/KAFKA-9203 + // detect buggy lz4 libraries on the classpath + static { + RuntimeException exception = null; + try { + detectBrokenLz4Version(); + } catch (RuntimeException e) { + exception = e; + } + BROKEN_LZ4_EXCEPTION = exception; + } + private final ByteBuffer in; private final boolean ignoreFlagDescriptorChecksum; private final BufferSupplier bufferSupplier; @@ -73,6 +87,9 @@ public final class KafkaLZ4BlockInputStream extends InputStream { * @throws IOException */ public KafkaLZ4BlockInputStream(ByteBuffer in, BufferSupplier bufferSupplier, boolean ignoreFlagDescriptorChecksum) throws IOException { + if (BROKEN_LZ4_EXCEPTION != null) { + throw BROKEN_LZ4_EXCEPTION; + } this.ignoreFlagDescriptorChecksum = ignoreFlagDescriptorChecksum; this.in = in.duplicate().order(ByteOrder.LITTLE_ENDIAN); this.bufferSupplier = bufferSupplier; @@ -263,4 +280,37 @@ public void reset() { public boolean markSupported() { return false; } + + /** + * Checks whether the version of lz4 on the classpath has the fix for reading from ByteBuffers with + * non-zero array offsets (see https://github.com/lz4/lz4-java/pull/65) + */ + static void detectBrokenLz4Version() { + byte[] source = new byte[]{1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 3, 3, 3, 3, 3}; + final LZ4Compressor compressor = LZ4Factory.fastestInstance().fastCompressor(); + + final byte[] compressed = new byte[compressor.maxCompressedLength(source.length)]; + final int compressedLength = compressor.compress(source, 0, source.length, compressed, 0, + compressed.length); + + // allocate an array-backed ByteBuffer with non-zero array-offset + final byte[] zeroes = {0, 0, 0, 0, 0}; + ByteBuffer nonZeroOffsetBuffer = ByteBuffer + .allocate(zeroes.length + compressed.length) + .put(zeroes) + .slice() + .put(compressed); + + ByteBuffer dest = ByteBuffer.allocate(source.length); + try { + DECOMPRESSOR.decompress(nonZeroOffsetBuffer, 0, compressedLength, dest, 0, source.length); + } catch (Exception e) { + throw new RuntimeException("Kafka has detected detected a buggy lz4-java library (< 1.4.x) on the classpath." + + " If you are using Kafka client libraries, make sure your application does not" + + " accidentally override the version provided by Kafka or include multiple versions" + + " of the library on the classpath. The lz4-java version on the classpath should" + + " match the version the Kafka client libraries depend on. Adding -verbose:class" + + " to your JVM arguments may help understand which lz4-java version is getting loaded.", e); + } + } } From 898e3e871e1f0e76cf663bda203f8ebded8335e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Xavier=20L=C3=A9aut=C3=A9?= Date: Thu, 22 Jul 2021 13:44:31 -0700 Subject: [PATCH 3/3] add comments --- .../common/compress/KafkaLZ4BlockInputStream.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java b/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java index d0ff033c764ac..e2fbd5ac04d98 100644 --- a/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java +++ b/clients/src/main/java/org/apache/kafka/common/compress/KafkaLZ4BlockInputStream.java @@ -293,13 +293,15 @@ static void detectBrokenLz4Version() { final int compressedLength = compressor.compress(source, 0, source.length, compressed, 0, compressed.length); - // allocate an array-backed ByteBuffer with non-zero array-offset + // allocate an array-backed ByteBuffer with non-zero array-offset containing the compressed data + // a buggy decompressor will read the data from the beginning of the underlying array instead of + // the beginning of the ByteBuffer, failing to decompress the invalid data. final byte[] zeroes = {0, 0, 0, 0, 0}; ByteBuffer nonZeroOffsetBuffer = ByteBuffer - .allocate(zeroes.length + compressed.length) - .put(zeroes) - .slice() - .put(compressed); + .allocate(zeroes.length + compressed.length) // allocates the backing array with extra space to offset the data + .put(zeroes) // prepend invalid bytes (zeros) before the compressed data in the array + .slice() // create a new ByteBuffer sharing the underlying array, offset to start on the compressed data + .put(compressed); // write the compressed data at the beginning of this new buffer ByteBuffer dest = ByteBuffer.allocate(source.length); try {