com.google.guava
@@ -246,33 +252,4 @@
-
-
-
- non-aarch64
-
-
- !aarch64
-
-
-
-
- jitpack.io
- https://jitpack.io
- Jitpack.io repository
-
-
-
-
-
- com.github.rdblue
- brotli-codec
- ${brotli-codec.version}
- runtime
- true
-
-
-
-
-
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java
index 9024954bb9..10b9960ef1 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/CodecFactory.java
@@ -41,6 +41,7 @@
import org.apache.parquet.compression.CompressionCodecFactory;
import org.apache.parquet.conf.HadoopParquetConfiguration;
import org.apache.parquet.conf.ParquetConfiguration;
+import org.apache.parquet.hadoop.codec.BrotliCodec;
import org.apache.parquet.hadoop.codec.Lz4RawCodec;
import org.apache.parquet.hadoop.codec.ZstandardCodec;
import org.apache.parquet.hadoop.metadata.CompressionCodecName;
@@ -178,7 +179,13 @@ public BytesInput decompress(BytesInput bytes, int decompressedSize) throws IOEx
if (decompressor != null) {
decompressor.reset();
}
- InputStream is = codec.createInputStream(bytes.toInputStream(), decompressor);
+ InputStream is;
+ if (codec instanceof BrotliCodec) {
+ // Brotli is decoded in one shot; bound the output by the size declared in the page header.
+ is = ((BrotliCodec) codec).createInputStream(bytes.toInputStream(), decompressedSize);
+ } else {
+ is = codec.createInputStream(bytes.toInputStream(), decompressor);
+ }
// Eagerly materialize the decompressed stream for codecs that require all input in a single buffer.
// ZSTD: releases off-heap resources early to avoid fragmentation (see parquet-format#398).
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java
index 4788c5f040..f51e0fa379 100644
--- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/DirectCodecFactory.java
@@ -136,6 +136,10 @@ protected BytesDecompressor createDecompressor(final CompressionCodecName codecN
if (codec == null) {
return NO_OP_DECOMPRESSOR;
}
+ if (codec.getDecompressorType() == null) {
+ // Stream-only codecs (for example Brotli) do not expose a Decompressor.
+ return new HeapBytesDecompressor(codec);
+ }
DirectCodecPool.CodecPool pool = DirectCodecPool.INSTANCE.codec(codec);
if (pool.supportsDirectDecompression()) {
return new FullDirectDecompressor(pool.borrowDirectDecompressor());
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCodec.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCodec.java
new file mode 100644
index 0000000000..d9d05960f7
--- /dev/null
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCodec.java
@@ -0,0 +1,151 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.parquet.hadoop.codec;
+
+import com.aayushatharva.brotli4j.Brotli4jLoader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionInputStream;
+import org.apache.hadoop.io.compress.CompressionOutputStream;
+import org.apache.hadoop.io.compress.Compressor;
+import org.apache.hadoop.io.compress.Decompressor;
+import org.apache.parquet.hadoop.BadConfigurationException;
+
+/**
+ * Hadoop codec for Brotli, backed by the maintained brotli4j native bindings.
+ *
+ * The previous implementation depended on the abandoned jbrotli-based brotli-codec artifact,
+ * which bundled an old Brotli decoder. This codec is loaded reflectively by {@link
+ * org.apache.parquet.hadoop.CodecFactory} and only requires brotli4j when Brotli is actually used.
+ */
+public class BrotliCodec implements CompressionCodec, Configurable {
+
+ public static final String BROTLI_QUALITY = "compression.brotli.quality";
+ public static final int DEFAULT_QUALITY = 1;
+ private static final int DEFAULT_MAX_OUTPUT_SIZE = Integer.MAX_VALUE - 8;
+
+ private Configuration conf;
+ private int quality = DEFAULT_QUALITY;
+
+ @Override
+ public void setConf(Configuration conf) {
+ this.conf = conf;
+ if (conf != null) {
+ quality = conf.getInt(BROTLI_QUALITY, DEFAULT_QUALITY);
+ }
+ }
+
+ @Override
+ public Configuration getConf() {
+ return conf;
+ }
+
+ @Override
+ public CompressionOutputStream createOutputStream(OutputStream out) throws IOException {
+ if (quality < 0 || quality > 11) {
+ throw new BadConfigurationException("Invalid Brotli compression quality: " + quality + ". Expected 0..11.");
+ }
+ ensureBrotliAvailable();
+ try {
+ return new BrotliCompressionOutputStream(out, quality);
+ } catch (LinkageError e) {
+ throw missingBrotli(e);
+ }
+ }
+
+ @Override
+ public CompressionOutputStream createOutputStream(OutputStream out, Compressor compressor) throws IOException {
+ return createOutputStream(out);
+ }
+
+ @Override
+ public Class getCompressorType() {
+ return null;
+ }
+
+ @Override
+ public Compressor createCompressor() {
+ return null;
+ }
+
+ /**
+ * Decodes a complete Brotli stream into memory. Callers that handle untrusted data should use
+ * {@link #createInputStream(InputStream, int)} so the decoded output is bounded by the size they
+ * expect.
+ */
+ @Override
+ public CompressionInputStream createInputStream(InputStream in) throws IOException {
+ return createInputStream(in, DEFAULT_MAX_OUTPUT_SIZE);
+ }
+
+ /**
+ * Creates a decoder with an upper bound on the decompressed size. Parquet passes the size from
+ * the page header, so a crafted page cannot expand without limit during decoding.
+ *
+ * @throws IOException if the bound is negative or the decoded output exceeds it
+ */
+ public CompressionInputStream createInputStream(InputStream in, int maxOutputSize) throws IOException {
+ ensureBrotliAvailable();
+ try {
+ return new BrotliCompressionInputStream(in, maxOutputSize);
+ } catch (LinkageError e) {
+ throw missingBrotli(e);
+ }
+ }
+
+ @Override
+ public CompressionInputStream createInputStream(InputStream in, Decompressor decompressor) throws IOException {
+ return createInputStream(in);
+ }
+
+ @Override
+ public Class getDecompressorType() {
+ return null;
+ }
+
+ @Override
+ public Decompressor createDecompressor() {
+ return null;
+ }
+
+ @Override
+ public String getDefaultExtension() {
+ return ".br";
+ }
+
+ private static void ensureBrotliAvailable() throws IOException {
+ try {
+ Brotli4jLoader.ensureAvailability();
+ } catch (LinkageError e) {
+ throw missingBrotli(e);
+ }
+ }
+
+ private static IOException missingBrotli(LinkageError cause) {
+ return new IOException(
+ "Brotli codec requires the optional brotli4j dependency and its native library "
+ + "for this platform on the classpath",
+ cause);
+ }
+}
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionInputStream.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionInputStream.java
new file mode 100644
index 0000000000..9cfa50a01e
--- /dev/null
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionInputStream.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.parquet.hadoop.codec;
+
+import com.aayushatharva.brotli4j.decoder.Decoder;
+import java.io.IOException;
+import java.io.InputStream;
+import org.apache.hadoop.io.compress.CompressionInputStream;
+
+class BrotliCompressionInputStream extends CompressionInputStream {
+
+ private final byte[] decompressed;
+ private int position;
+
+ BrotliCompressionInputStream(InputStream in, int maxOutputSize) throws IOException {
+ super(in);
+ if (maxOutputSize < 0) {
+ throw new IOException(
+ "Invalid Brotli maximum output size: " + maxOutputSize + ". Expected a non-negative size.");
+ }
+ byte[] compressed = in.readAllBytes();
+ if (compressed.length == 0) {
+ this.decompressed = new byte[0];
+ return;
+ }
+ // The native decoder requires a positive limit, including for an empty decoded stream.
+ int outputLimit = Math.max(maxOutputSize, 1);
+ try {
+ // brotli4j's streaming BrotliInputStream fails on valid streams whose output crosses the
+ // encoder window (hyperxpro/Brotli4j#267). Parquet feeds one page at a time, so use the
+ // bounded one-shot decoder, which handles those streams without allowing unbounded output.
+ byte[] decoded = Decoder.decompress(compressed, 0, compressed.length, outputLimit);
+ if (decoded.length > maxOutputSize) {
+ throw new IOException(
+ "Brotli decompressed size " + decoded.length + " exceeds maximum size " + maxOutputSize);
+ }
+ this.decompressed = decoded;
+ } catch (IOException | RuntimeException e) {
+ throw new IOException("Failed to decompress Brotli stream", e);
+ }
+ }
+
+ @Override
+ public int read() {
+ if (position >= decompressed.length) {
+ return -1;
+ }
+ return decompressed[position++] & 0xFF;
+ }
+
+ @Override
+ public int read(byte[] b, int off, int len) {
+ if (len == 0) {
+ return 0;
+ }
+ if (position >= decompressed.length) {
+ return -1;
+ }
+ int count = Math.min(len, decompressed.length - position);
+ System.arraycopy(decompressed, position, b, off, count);
+ position += count;
+ return count;
+ }
+
+ @Override
+ public int available() {
+ return decompressed.length - position;
+ }
+
+ @Override
+ public void resetState() throws IOException {
+ throw new IOException("BrotliCompressionInputStream does not support resetState");
+ }
+
+ @Override
+ public void seek(long pos) {
+ if (pos < 0 || pos > decompressed.length) {
+ throw new IllegalArgumentException("Invalid position: " + pos);
+ }
+ position = (int) pos;
+ }
+
+ @Override
+ public long getPos() {
+ return position;
+ }
+}
diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionOutputStream.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionOutputStream.java
new file mode 100644
index 0000000000..8783b77ff6
--- /dev/null
+++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionOutputStream.java
@@ -0,0 +1,170 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.parquet.hadoop.codec;
+
+import com.aayushatharva.brotli4j.encoder.BrotliOutputStream;
+import com.aayushatharva.brotli4j.encoder.Encoder;
+import java.io.FilterOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import org.apache.hadoop.io.compress.CompressionOutputStream;
+
+class BrotliCompressionOutputStream extends CompressionOutputStream {
+
+ private final BrotliOutputStream encoder;
+ private boolean finished;
+ private boolean closed;
+ private IOException failure;
+
+ BrotliCompressionOutputStream(OutputStream out, int quality) throws IOException {
+ super(out);
+ this.encoder = new BrotliOutputStream(new NonClosingOutputStream(out), Encoder.Parameters.create(quality));
+ }
+
+ @Override
+ public void write(int b) throws IOException {
+ ensureWritable();
+ try {
+ encoder.write(b);
+ } catch (IOException e) {
+ failure = e;
+ throw e;
+ }
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ ensureWritable();
+ try {
+ encoder.write(b, off, len);
+ } catch (IOException e) {
+ failure = e;
+ throw e;
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ ensureNoFailure();
+ try {
+ if (!finished) {
+ encoder.flush();
+ }
+ out.flush();
+ } catch (IOException e) {
+ failure = e;
+ throw e;
+ }
+ }
+
+ @Override
+ public void finish() throws IOException {
+ ensureNoFailure();
+ finishEncoder();
+ }
+
+ private void finishEncoder() throws IOException {
+ if (finished) {
+ return;
+ }
+ try {
+ encoder.close();
+ finished = true;
+ } catch (IOException e) {
+ if (failure == null) {
+ failure = e;
+ }
+ throw e;
+ }
+ }
+
+ @Override
+ public void resetState() throws IOException {
+ throw new IOException("BrotliCompressionOutputStream does not support resetState");
+ }
+
+ @Override
+ public void close() throws IOException {
+ if (closed) {
+ return;
+ }
+ // Use a distinct exception so try-with-resources can suppress it onto the original failure.
+ IOException closeFailure =
+ failure == null ? null : new IOException("Previous Brotli stream operation failed", failure);
+ try {
+ // Release native resources even when a previous write or flush has failed.
+ finishEncoder();
+ } catch (IOException e) {
+ if (closeFailure == null) {
+ closeFailure = e;
+ } else if (closeFailure != e && closeFailure.getCause() != e) {
+ closeFailure.addSuppressed(e);
+ }
+ }
+ try {
+ out.close();
+ } catch (IOException e) {
+ if (closeFailure == null) {
+ closeFailure = e;
+ } else if (closeFailure != e && closeFailure.getCause() != e) {
+ closeFailure.addSuppressed(e);
+ }
+ }
+ closed = true;
+ if (closeFailure != null) {
+ throw closeFailure;
+ }
+ }
+
+ private void ensureWritable() throws IOException {
+ ensureNoFailure();
+ if (finished) {
+ throw new IOException("Stream is finished");
+ }
+ }
+
+ private void ensureNoFailure() throws IOException {
+ if (failure != null) {
+ throw new IOException("Previous Brotli stream operation failed", failure);
+ }
+ }
+
+ /**
+ * BrotliOutputStream closes its target when it finalizes the Brotli stream. Hadoop finishes a
+ * codec stream without closing the caller's target, so swallow close here and only flush. The
+ * codec's own close() closes the target explicitly.
+ */
+ private static final class NonClosingOutputStream extends FilterOutputStream {
+
+ NonClosingOutputStream(OutputStream out) {
+ super(out);
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ out.write(b, off, len);
+ }
+
+ @Override
+ public void close() throws IOException {
+ flush();
+ }
+ }
+}
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java
index 1d9171897f..6db09cde2b 100644
--- a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestDirectCodecFactory.java
@@ -17,7 +17,6 @@
*/
package org.apache.parquet.hadoop;
-import static org.apache.parquet.hadoop.metadata.CompressionCodecName.BROTLI;
import static org.apache.parquet.hadoop.metadata.CompressionCodecName.GZIP;
import static org.apache.parquet.hadoop.metadata.CompressionCodecName.LZ4;
import static org.apache.parquet.hadoop.metadata.CompressionCodecName.LZ4_RAW;
@@ -216,11 +215,6 @@ public void compressionCodecs() {
Set codecsToSkip = new HashSet<>();
codecsToSkip.add(LZO); // not distributed because it is GPL
codecsToSkip.add(LZ4); // not distributed in the default version of Hadoop
- final String arch = System.getProperty("os.arch");
- if ("aarch64".equals(arch)) {
- // PARQUET-1975 brotli-codec does not have natives for ARM64 architectures
- codecsToSkip.add(BROTLI);
- }
for (final int size : sizes) {
for (final boolean useOnHeapComp : comp) {
diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/codec/TestBrotliCodec.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/codec/TestBrotliCodec.java
new file mode 100644
index 0000000000..e7d2375c97
--- /dev/null
+++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/codec/TestBrotliCodec.java
@@ -0,0 +1,407 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.parquet.hadoop.codec;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import com.aayushatharva.brotli4j.encoder.BrotliOutputStream;
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.OutputStream;
+import java.lang.reflect.Field;
+import java.nio.charset.StandardCharsets;
+import java.util.Random;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.compress.CompressionInputStream;
+import org.apache.hadoop.io.compress.CompressionOutputStream;
+import org.apache.parquet.bytes.BytesInput;
+import org.apache.parquet.hadoop.BadConfigurationException;
+import org.apache.parquet.hadoop.CodecFactory;
+import org.apache.parquet.hadoop.metadata.CompressionCodecName;
+import org.junit.jupiter.api.Test;
+
+public class TestBrotliCodec {
+
+ @Test
+ public void testRoundTrip() throws Exception {
+ byte[] input = new byte[64 * 1024];
+ new Random(42).nextBytes(input);
+
+ BrotliCodec codec = newCodec(5);
+ try (CompressionInputStream in = codec.createInputStream(compress(codec, input))) {
+ assertThat(in.readAllBytes()).isEqualTo(input);
+ }
+ }
+
+ @Test
+ public void testLargeStreamCrossingEncoderWindow() throws Exception {
+ // brotli4j's streaming decoder fails once the decompressed output crosses the 4 MiB encoder
+ // window (hyperxpro/Brotli4j#267). Parquet pages can be larger than that, so keep a regression
+ // test for the one-shot decoder path.
+ byte[] input = largeCompressibleInput();
+ BrotliCodec codec = newCodec(5);
+
+ try (CompressionInputStream in = codec.createInputStream(compress(codec, input))) {
+ assertThat(in.readAllBytes()).isEqualTo(input);
+ }
+ }
+
+ @Test
+ public void testLargeStreamThroughCodecFactory() throws Exception {
+ byte[] input = largeCompressibleInput();
+ CodecFactory factory = new CodecFactory(new Configuration(false), 64 * 1024);
+ try {
+ BytesInput compressed =
+ factory.getCompressor(CompressionCodecName.BROTLI, 5).compress(BytesInput.from(input));
+ BytesInput decompressed =
+ factory.getDecompressor(CompressionCodecName.BROTLI).decompress(compressed, input.length);
+ assertThat(decompressed.toByteArray()).isEqualTo(input);
+ } finally {
+ factory.release();
+ }
+ }
+
+ @Test
+ public void testCloseClosesTarget() throws Exception {
+ TrackingOutputStream target = new TrackingOutputStream();
+ BrotliCodec codec = newCodec(5);
+ try (CompressionOutputStream out = codec.createOutputStream(target)) {
+ out.write(new byte[] {1, 2, 3});
+ }
+ assertThat(target.closed).isTrue();
+ }
+
+ @Test
+ public void testFlushThenFinishRoundTrips() throws Exception {
+ byte[] input = new byte[1024 * 1024];
+ new Random(11).nextBytes(input);
+ BrotliCodec codec = newCodec(5);
+ TrackingOutputStream target = new TrackingOutputStream();
+
+ try (CompressionOutputStream out = codec.createOutputStream(target)) {
+ out.write(input, 0, input.length / 2);
+ out.flush();
+ assertThat(target.flushes).isGreaterThan(0);
+ assertThat(target.size()).isGreaterThan(0);
+
+ out.write(input, input.length / 2, input.length / 2);
+ out.finish();
+ }
+
+ try (CompressionInputStream in = codec.createInputStream(new ByteArrayInputStream(target.toByteArray()))) {
+ assertThat(in.readAllBytes()).isEqualTo(input);
+ }
+ }
+
+ @Test
+ public void testBoundedDecompressionRejectsBomb() throws Exception {
+ byte[] zeros = new byte[16 * 1024 * 1024];
+ byte[] compressed = compressBytes(newCodec(5), zeros);
+
+ assertThatThrownBy(() -> newCodec(5).createInputStream(new ByteArrayInputStream(compressed), 1024))
+ .isInstanceOf(IOException.class)
+ .hasMessage("Failed to decompress Brotli stream")
+ .hasStackTraceContaining("exceeds maximum size");
+
+ CodecFactory factory = new CodecFactory(new Configuration(false), 64 * 1024);
+ try {
+ assertThatThrownBy(() -> factory.getDecompressor(CompressionCodecName.BROTLI)
+ .decompress(BytesInput.from(compressed), 1024))
+ .isInstanceOf(IOException.class)
+ .hasMessage("Failed to decompress Brotli stream")
+ .hasStackTraceContaining("exceeds maximum size");
+ } finally {
+ factory.release();
+ }
+ }
+
+ @Test
+ public void testInvalidQualityOnlyRejectsCompression() throws Exception {
+ byte[] input = new byte[1024];
+ new Random(13).nextBytes(input);
+ byte[] compressed = compressBytes(newCodec(5), input);
+
+ BrotliCodec codec = newCodec(99);
+ try (CompressionInputStream in = codec.createInputStream(new ByteArrayInputStream(compressed), input.length)) {
+ assertThat(in.readAllBytes()).isEqualTo(input);
+ }
+
+ assertThatThrownBy(() -> codec.createOutputStream(new ByteArrayOutputStream()))
+ .isInstanceOf(BadConfigurationException.class)
+ .hasMessage("Invalid Brotli compression quality: 99. Expected 0..11.");
+ }
+
+ @Test
+ public void testEmptyAndZeroSizeStreams() throws Exception {
+ byte[] compressed = compressBytes(newCodec(5), new byte[0]);
+ try (CompressionInputStream in = newCodec(5).createInputStream(new ByteArrayInputStream(compressed), 0)) {
+ assertThat(in.readAllBytes()).isEmpty();
+ }
+ }
+
+ @Test
+ public void testZeroSizeRejectsOneByteOutput() throws Exception {
+ byte[] compressed = compressBytes(newCodec(5), new byte[] {1});
+
+ assertThatThrownBy(() -> newCodec(5).createInputStream(new ByteArrayInputStream(compressed), 0))
+ .isInstanceOf(IOException.class)
+ .hasMessage("Failed to decompress Brotli stream")
+ .hasRootCauseMessage("Brotli decompressed size 1 exceeds maximum size 0");
+
+ CodecFactory factory = new CodecFactory(new Configuration(false), 64 * 1024);
+ try {
+ assertThatThrownBy(() -> factory.getDecompressor(CompressionCodecName.BROTLI)
+ .decompress(BytesInput.from(compressed), 0))
+ .isInstanceOf(IOException.class)
+ .hasMessage("Failed to decompress Brotli stream")
+ .hasRootCauseMessage("Brotli decompressed size 1 exceeds maximum size 0");
+ } finally {
+ factory.release();
+ }
+ }
+
+ @Test
+ public void testNegativeSizeIsRejectedForEmptyAndNonEmptyInput() throws Exception {
+ byte[][] inputs = {new byte[0], compressBytes(newCodec(5), new byte[] {1})};
+ for (byte[] input : inputs) {
+ assertThatThrownBy(() -> newCodec(5).createInputStream(new ByteArrayInputStream(input), -1))
+ .isInstanceOf(IOException.class)
+ .hasMessage("Invalid Brotli maximum output size: -1. Expected a non-negative size.");
+ }
+ }
+
+ @Test
+ public void testCloseAfterFlushFailureClosesEncoderAndTarget() throws Exception {
+ FailingOutputStream target = new FailingOutputStream();
+ CompressionOutputStream out = newCodec(5).createOutputStream(target);
+ BrotliOutputStream encoder = encoderOf(out);
+ out.write(new byte[] {1, 2, 3});
+ target.failOnFlush = true;
+
+ assertThatThrownBy(out::flush).isInstanceOf(IOException.class).hasMessage("flush failed");
+ assertThatThrownBy(out::close).isInstanceOf(IOException.class).hasMessageContaining("failed");
+ assertThatThrownBy(encoder::flush).isInstanceOf(IOException.class).hasMessage("write after close");
+ out.close();
+ assertThat(target.closes).isEqualTo(1);
+ }
+
+ @Test
+ public void testCloseAfterWriteFailureClosesEncoderAndTarget() throws Exception {
+ FailingOutputStream target = new FailingOutputStream();
+ CompressionOutputStream out = newCodec(1).createOutputStream(target);
+ BrotliOutputStream encoder = encoderOf(out);
+ byte[] input = new byte[1024 * 1024];
+ new Random(17).nextBytes(input);
+ target.failOnWrite = true;
+
+ assertThatThrownBy(() -> out.write(input))
+ .isInstanceOf(IOException.class)
+ .hasMessage("write failed");
+ assertThatThrownBy(out::close).isInstanceOf(IOException.class).hasMessageContaining("failed");
+ assertThatThrownBy(encoder::flush).isInstanceOf(IOException.class).hasMessage("write after close");
+ out.close();
+ assertThat(target.closes).isEqualTo(1);
+ }
+
+ @Test
+ public void testClosePreservesFailureWhenCleanupAlsoFails() throws Exception {
+ FailingOutputStream target = new FailingOutputStream();
+ CompressionOutputStream out = newCodec(5).createOutputStream(target);
+ out.write(new byte[] {1, 2, 3});
+ target.failOnFlush = true;
+ target.failOnClose = true;
+
+ assertThatThrownBy(out::flush).isInstanceOf(IOException.class).hasMessage("flush failed");
+ assertThatThrownBy(out::close)
+ .isInstanceOf(IOException.class)
+ .hasMessage("Previous Brotli stream operation failed")
+ .hasCause(target.flushFailure)
+ .hasSuppressedException(target.closeFailure);
+ out.close();
+ assertThat(target.closes).isEqualTo(1);
+ }
+
+ @Test
+ public void testTryWithResourcesPreservesWriteFailure() throws Exception {
+ FailingOutputStream target = new FailingOutputStream();
+ byte[] input = new byte[1024 * 1024];
+ new Random(19).nextBytes(input);
+ target.failOnWrite = true;
+
+ assertThatThrownBy(() -> {
+ try (CompressionOutputStream out = newCodec(1).createOutputStream(target)) {
+ out.write(input);
+ }
+ })
+ .isInstanceOf(IOException.class)
+ .hasMessage("write failed")
+ .isSameAs(target.writeFailure);
+ assertThat(target.writeFailure.getSuppressed()).hasSize(1);
+ assertThat(target.writeFailure.getSuppressed()[0])
+ .isInstanceOf(IOException.class)
+ .hasMessage("Previous Brotli stream operation failed")
+ .hasCause(target.writeFailure);
+ assertThat(target.closes).isEqualTo(1);
+ }
+
+ @Test
+ public void testCloseAfterFinishFailureIsIdempotent() throws Exception {
+ FailingOutputStream target = new FailingOutputStream();
+ BrotliCodec codec = newCodec(5);
+ CompressionOutputStream out = codec.createOutputStream(target);
+ out.write(new byte[100 * 1024]);
+ target.failOnFlush = true;
+
+ assertThatThrownBy(out::finish).isInstanceOf(IOException.class).hasMessage("flush failed");
+ assertThatThrownBy(out::close).isInstanceOf(IOException.class).hasMessageContaining("failed");
+ out.close();
+ assertThat(target.closes).isEqualTo(1);
+ }
+
+ @Test
+ public void testFinishFailureIsReportedOnRetry() throws Exception {
+ FailingOutputStream target = new FailingOutputStream();
+ BrotliCodec codec = newCodec(5);
+ CompressionOutputStream out = codec.createOutputStream(target);
+ out.write(new byte[100 * 1024]);
+ target.failOnFlush = true;
+
+ assertThatThrownBy(out::finish).isInstanceOf(IOException.class).hasMessage("flush failed");
+ assertThatThrownBy(out::finish)
+ .isInstanceOf(IOException.class)
+ .hasMessage("Previous Brotli stream operation failed")
+ .hasCause(target.flushFailure);
+ assertThatThrownBy(out::close).isInstanceOf(IOException.class).hasMessageContaining("failed");
+ }
+
+ @Test
+ public void testPositionAndSeek() throws Exception {
+ byte[] input = new byte[100];
+ new Random(7).nextBytes(input);
+ BrotliCodec codec = newCodec(5);
+
+ try (CompressionInputStream in = codec.createInputStream(compress(codec, input))) {
+ assertThat(in.read()).isEqualTo(input[0] & 0xFF);
+ assertThat(in.getPos()).isEqualTo(1);
+
+ in.seek(0);
+ assertThat(in.getPos()).isZero();
+ assertThat(in.readAllBytes()).isEqualTo(input);
+ assertThatThrownBy(in::resetState)
+ .isInstanceOf(IOException.class)
+ .hasMessage("BrotliCompressionInputStream does not support resetState");
+ }
+ }
+
+ private static BrotliCodec newCodec(int quality) {
+ Configuration conf = new Configuration(false);
+ conf.setInt(BrotliCodec.BROTLI_QUALITY, quality);
+ BrotliCodec codec = new BrotliCodec();
+ codec.setConf(conf);
+ return codec;
+ }
+
+ private static BrotliOutputStream encoderOf(CompressionOutputStream out) throws ReflectiveOperationException {
+ // Verify the native encoder's lifecycle without exposing it through the production codec API.
+ Field field = BrotliCompressionOutputStream.class.getDeclaredField("encoder");
+ field.setAccessible(true);
+ return (BrotliOutputStream) field.get(out);
+ }
+
+ private static ByteArrayInputStream compress(BrotliCodec codec, byte[] input) throws IOException {
+ return new ByteArrayInputStream(compressBytes(codec, input));
+ }
+
+ private static byte[] compressBytes(BrotliCodec codec, byte[] input) throws IOException {
+ ByteArrayOutputStream compressed = new ByteArrayOutputStream();
+ try (CompressionOutputStream out = codec.createOutputStream(compressed)) {
+ out.write(input);
+ out.finish();
+ }
+ return compressed.toByteArray();
+ }
+
+ private static byte[] largeCompressibleInput() {
+ StringBuilder builder = new StringBuilder();
+ String line = "at com.example.SomeClass.someMethod(SomeClass.java:123)\n";
+ while (builder.length() < 5 * 1024 * 1024) {
+ builder.append(line);
+ }
+ return builder.toString().getBytes(StandardCharsets.UTF_8);
+ }
+
+ private static final class TrackingOutputStream extends ByteArrayOutputStream {
+ private boolean closed;
+ private int flushes;
+
+ @Override
+ public void close() throws IOException {
+ closed = true;
+ super.close();
+ }
+
+ @Override
+ public void flush() throws IOException {
+ flushes++;
+ super.flush();
+ }
+ }
+
+ private static final class FailingOutputStream extends OutputStream {
+ private final IOException writeFailure = new IOException("write failed");
+ private final IOException flushFailure = new IOException("flush failed");
+ private final IOException closeFailure = new IOException("close failed");
+ private boolean failOnWrite;
+ private boolean failOnFlush;
+ private boolean failOnClose;
+ private int closes;
+
+ @Override
+ public void write(int b) throws IOException {
+ if (failOnWrite) {
+ throw writeFailure;
+ }
+ }
+
+ @Override
+ public void write(byte[] b, int off, int len) throws IOException {
+ if (failOnWrite) {
+ throw writeFailure;
+ }
+ }
+
+ @Override
+ public void flush() throws IOException {
+ if (failOnFlush) {
+ throw flushFailure;
+ }
+ }
+
+ @Override
+ public void close() throws IOException {
+ closes++;
+ if (failOnClose) {
+ throw closeFailure;
+ }
+ }
+ }
+}
diff --git a/pom.xml b/pom.xml
index 8564229b44..5064390172 100644
--- a/pom.xml
+++ b/pom.xml
@@ -95,7 +95,7 @@
1.7.33
1.12.2
33.7.2-jre
- 0.1.1
+ 1.24.0
3.27.7
6.1.3
3.27.7