From a4d4dc18bc4307d9ee1cd6cc632cf285898e2808 Mon Sep 17 00:00:00 2001 From: Donghun Ha Date: Sun, 25 Oct 2026 01:43:26 +0900 Subject: [PATCH 1/6] Replace jbrotli Brotli codec with brotli4j for CVE-2026-111497 --- .../hadoop/metadata/CompressionCodecName.java | 2 +- parquet-hadoop/pom.xml | 35 +----- .../parquet/hadoop/DirectCodecFactory.java | 4 + .../parquet/hadoop/codec/BrotliCodec.java | 115 ++++++++++++++++++ .../codec/BrotliCompressionInputStream.java | 65 ++++++++++ .../codec/BrotliCompressionOutputStream.java | 95 +++++++++++++++ .../hadoop/TestDirectCodecFactory.java | 6 - .../parquet/hadoop/codec/TestBrotliCodec.java | 55 +++++++++ pom.xml | 2 +- 9 files changed, 342 insertions(+), 37 deletions(-) create mode 100644 parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCodec.java create mode 100644 parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionInputStream.java create mode 100644 parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionOutputStream.java create mode 100644 parquet-hadoop/src/test/java/org/apache/parquet/hadoop/codec/TestBrotliCodec.java diff --git a/parquet-common/src/main/java/org/apache/parquet/hadoop/metadata/CompressionCodecName.java b/parquet-common/src/main/java/org/apache/parquet/hadoop/metadata/CompressionCodecName.java index 18818c3c21..5354e6c690 100644 --- a/parquet-common/src/main/java/org/apache/parquet/hadoop/metadata/CompressionCodecName.java +++ b/parquet-common/src/main/java/org/apache/parquet/hadoop/metadata/CompressionCodecName.java @@ -27,7 +27,7 @@ public enum CompressionCodecName { SNAPPY("org.apache.parquet.hadoop.codec.SnappyCodec", CompressionCodec.SNAPPY, ".snappy"), GZIP("org.apache.hadoop.io.compress.GzipCodec", CompressionCodec.GZIP, ".gz"), LZO("com.hadoop.compression.lzo.LzoCodec", CompressionCodec.LZO, ".lzo"), - BROTLI("org.apache.hadoop.io.compress.BrotliCodec", CompressionCodec.BROTLI, ".br"), + BROTLI("org.apache.parquet.hadoop.codec.BrotliCodec", CompressionCodec.BROTLI, ".br"), LZ4("org.apache.hadoop.io.compress.Lz4Codec", CompressionCodec.LZ4, ".lz4hadoop"), ZSTD("org.apache.parquet.hadoop.codec.ZstandardCodec", CompressionCodec.ZSTD, ".zstd"), LZ4_RAW("org.apache.parquet.hadoop.codec.Lz4RawCodec", CompressionCodec.LZ4_RAW, ".lz4raw"); diff --git a/parquet-hadoop/pom.xml b/parquet-hadoop/pom.xml index f2997a1ae9..3a24bb673a 100644 --- a/parquet-hadoop/pom.xml +++ b/parquet-hadoop/pom.xml @@ -174,6 +174,12 @@ zstd-jni ${zstd-jni.version} + + com.aayushatharva.brotli4j + brotli4j + ${brotli4j.version} + true + 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/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..54cb8b9e7c --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCodec.java @@ -0,0 +1,115 @@ +/* + * 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; + +/** + * 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 = 5; + + private Configuration conf; + private int quality = DEFAULT_QUALITY; + + @Override + public void setConf(Configuration conf) { + this.conf = conf; + if (conf != null) { + int configuredQuality = conf.getInt(BROTLI_QUALITY, DEFAULT_QUALITY); + if (configuredQuality < 0 || configuredQuality > 11) { + throw new IllegalArgumentException( + "Invalid Brotli compression quality: " + configuredQuality + ". Expected 0..11."); + } + quality = configuredQuality; + } + } + + @Override + public Configuration getConf() { + return conf; + } + + @Override + public CompressionOutputStream createOutputStream(OutputStream out) throws IOException { + Brotli4jLoader.ensureAvailability(); + return new BrotliCompressionOutputStream(out, quality); + } + + @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; + } + + @Override + public CompressionInputStream createInputStream(InputStream in) throws IOException { + Brotli4jLoader.ensureAvailability(); + return new BrotliCompressionInputStream(in); + } + + @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"; + } +} 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..91fa40da8c --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionInputStream.java @@ -0,0 +1,65 @@ +/* + * 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.BrotliInputStream; +import java.io.IOException; +import java.io.InputStream; +import org.apache.hadoop.io.compress.CompressionInputStream; + +class BrotliCompressionInputStream extends CompressionInputStream { + + private final BrotliInputStream in; + + BrotliCompressionInputStream(InputStream in) throws IOException { + super(in); + this.in = new BrotliInputStream(in); + } + + @Override + public int read() throws IOException { + return in.read(); + } + + @Override + public int read(byte[] b, int off, int len) throws IOException { + return in.read(b, off, len); + } + + @Override + public void resetState() throws IOException { + throw new IOException("BrotliCompressionInputStream does not support resetState"); + } + + @Override + public void seek(long pos) { + throw new UnsupportedOperationException("BrotliCompressionInputStream does not support seek"); + } + + @Override + public long getPos() { + return 0; + } + + @Override + public void close() throws IOException { + in.close(); + } +} 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..6777df5262 --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCompressionOutputStream.java @@ -0,0 +1,95 @@ +/* + * 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 out; + private boolean finished; + + BrotliCompressionOutputStream(OutputStream out, int quality) throws IOException { + super(out); + this.out = new BrotliOutputStream(new NonClosingOutputStream(out), Encoder.Parameters.create(quality)); + } + + @Override + public void write(int b) throws IOException { + ensureOpen(); + out.write(b); + } + + @Override + public void write(byte[] b, int off, int len) throws IOException { + ensureOpen(); + out.write(b, off, len); + } + + @Override + public void finish() throws IOException { + if (!finished) { + out.close(); + finished = true; + } + } + + @Override + public void resetState() throws IOException { + throw new IOException("BrotliCompressionOutputStream does not support resetState"); + } + + @Override + public void close() throws IOException { + finish(); + } + + private void ensureOpen() throws IOException { + if (finished) { + throw new IOException("Stream is finished"); + } + } + + /** + * 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. + */ + 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..2733321d0f --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/codec/TestBrotliCodec.java @@ -0,0 +1,55 @@ +/* + * 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.junit.jupiter.api.Assertions.assertArrayEquals; + +import java.util.Random; +import org.apache.hadoop.conf.Configuration; +import org.apache.parquet.bytes.BytesInput; +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); + + Configuration conf = new Configuration(false); + conf.setInt(BrotliCodec.BROTLI_QUALITY, 5); + + CodecFactory factory = new CodecFactory(conf, 64 * 1024); + try { + CodecFactory.BytesCompressor compressor = factory.getCompressor(CompressionCodecName.BROTLI); + BytesInput compressed = compressor.compress(BytesInput.from(input)); + + CodecFactory.BytesDecompressor decompressor = + factory.getDecompressor(CompressionCodecName.BROTLI); + BytesInput decompressed = decompressor.decompress(compressed, input.length); + + assertArrayEquals(input, decompressed.toByteArray()); + } finally { + factory.release(); + } + } +} 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 From 22b938de881be8712cb8d44eb18402e90f3682c4 Mon Sep 17 00:00:00 2001 From: Donghun Ha Date: Tue, 27 Oct 2026 12:13:11 +0900 Subject: [PATCH 2/6] Fix Brotli codec stream handling and large-page decoding --- .../parquet/hadoop/codec/BrotliCodec.java | 33 ++++- .../codec/BrotliCompressionInputStream.java | 54 +++++--- .../codec/BrotliCompressionOutputStream.java | 71 ++++++++-- .../parquet/hadoop/codec/TestBrotliCodec.java | 126 ++++++++++++++++-- 4 files changed, 242 insertions(+), 42 deletions(-) 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 index 54cb8b9e7c..e146e566b0 100644 --- 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 @@ -41,7 +41,7 @@ public class BrotliCodec implements CompressionCodec, Configurable { public static final String BROTLI_QUALITY = "compression.brotli.quality"; - public static final int DEFAULT_QUALITY = 5; + public static final int DEFAULT_QUALITY = 1; private Configuration conf; private int quality = DEFAULT_QUALITY; @@ -66,8 +66,12 @@ public Configuration getConf() { @Override public CompressionOutputStream createOutputStream(OutputStream out) throws IOException { - Brotli4jLoader.ensureAvailability(); - return new BrotliCompressionOutputStream(out, quality); + ensureBrotliAvailable(); + try { + return new BrotliCompressionOutputStream(out, quality); + } catch (LinkageError e) { + throw missingBrotli(e); + } } @Override @@ -88,8 +92,12 @@ public Compressor createCompressor() { @Override public CompressionInputStream createInputStream(InputStream in) throws IOException { - Brotli4jLoader.ensureAvailability(); - return new BrotliCompressionInputStream(in); + ensureBrotliAvailable(); + try { + return new BrotliCompressionInputStream(in); + } catch (LinkageError e) { + throw missingBrotli(e); + } } @Override @@ -112,4 +120,19 @@ public Decompressor createDecompressor() { 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 index 91fa40da8c..72284ceddc 100644 --- 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 @@ -19,47 +19,71 @@ package org.apache.parquet.hadoop.codec; -import com.aayushatharva.brotli4j.decoder.BrotliInputStream; +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 BrotliInputStream in; + private final byte[] decompressed; + private int position; BrotliCompressionInputStream(InputStream in) throws IOException { super(in); - this.in = new BrotliInputStream(in); + byte[] compressed = in.readAllBytes(); + 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 + // one-shot decoder, which handles those streams correctly. + this.decompressed = Decoder.decompress(compressed, 0, compressed.length); + } catch (IOException e) { + throw new IOException("Failed to decompress Brotli stream", e); + } } @Override - public int read() throws IOException { - return in.read(); + public int read() { + if (position >= decompressed.length) { + return -1; + } + return decompressed[position++] & 0xFF; } @Override - public int read(byte[] b, int off, int len) throws IOException { - return in.read(b, off, len); + 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 void resetState() throws IOException { - throw new IOException("BrotliCompressionInputStream does not support resetState"); + public int available() { + return decompressed.length - position; } @Override - public void seek(long pos) { - throw new UnsupportedOperationException("BrotliCompressionInputStream does not support seek"); + public void resetState() { + position = 0; } @Override - public long getPos() { - return 0; + public void seek(long pos) { + if (pos < 0 || pos > decompressed.length) { + throw new IllegalArgumentException("Invalid position: " + pos); + } + position = (int) pos; } @Override - public void close() throws IOException { - in.close(); + 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 index 6777df5262..411711bdd8 100644 --- 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 @@ -28,31 +28,51 @@ class BrotliCompressionOutputStream extends CompressionOutputStream { - private final BrotliOutputStream out; + private final BrotliOutputStream encoder; private boolean finished; + private boolean closed; + private IOException failure; BrotliCompressionOutputStream(OutputStream out, int quality) throws IOException { super(out); - this.out = new BrotliOutputStream(new NonClosingOutputStream(out), Encoder.Parameters.create(quality)); + this.encoder = + new BrotliOutputStream(new NonClosingOutputStream(out), Encoder.Parameters.create(quality)); } @Override public void write(int b) throws IOException { - ensureOpen(); - out.write(b); + ensureWritable(); + encoder.write(b); } @Override public void write(byte[] b, int off, int len) throws IOException { - ensureOpen(); - out.write(b, off, len); + ensureWritable(); + encoder.write(b, off, len); + } + + @Override + public void flush() throws IOException { + ensureNoFailure(); + if (finished) { + out.flush(); + return; + } + encoder.flush(); } @Override public void finish() throws IOException { - if (!finished) { - out.close(); + ensureNoFailure(); + if (finished) { + return; + } + try { + encoder.close(); finished = true; + } catch (IOException e) { + failure = e; + throw e; } } @@ -63,18 +83,47 @@ public void resetState() throws IOException { @Override public void close() throws IOException { - finish(); + if (closed) { + return; + } + IOException closeFailure = null; + try { + finish(); + } catch (IOException e) { + closeFailure = e; + } + try { + out.close(); + } catch (IOException e) { + if (closeFailure == null) { + closeFailure = e; + } else { + closeFailure.addSuppressed(e); + } + } + if (closeFailure != null) { + throw closeFailure; + } + closed = true; } - private void ensureOpen() throws IOException { + 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. + * 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 { 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 index 2733321d0f..c2df6d8d13 100644 --- 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 @@ -19,10 +19,16 @@ package org.apache.parquet.hadoop.codec; -import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.assertj.core.api.Assertions.assertThat; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +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.CodecFactory; import org.apache.parquet.hadoop.metadata.CompressionCodecName; @@ -35,21 +41,119 @@ public void testRoundTrip() throws Exception { byte[] input = new byte[64 * 1024]; new Random(42).nextBytes(input); - Configuration conf = new Configuration(false); - conf.setInt(BrotliCodec.BROTLI_QUALITY, 5); + BrotliCodec codec = newCodec(5); + try (CompressionInputStream in = codec.createInputStream(compress(codec, input))) { + assertThat(in.readAllBytes()).isEqualTo(input); + } + } - CodecFactory factory = new CodecFactory(conf, 64 * 1024); - try { - CodecFactory.BytesCompressor compressor = factory.getCompressor(CompressionCodecName.BROTLI); - BytesInput compressed = compressor.compress(BytesInput.from(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); - CodecFactory.BytesDecompressor decompressor = - factory.getDecompressor(CompressionCodecName.BROTLI); - BytesInput decompressed = decompressor.decompress(compressed, input.length); + try (CompressionInputStream in = codec.createInputStream(compress(codec, input))) { + assertThat(in.readAllBytes()).isEqualTo(input); + } + } - assertArrayEquals(input, decompressed.toByteArray()); + @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).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); + ByteArrayOutputStream compressed = new ByteArrayOutputStream(); + + try (CompressionOutputStream out = codec.createOutputStream(compressed)) { + out.write(input, 0, input.length / 2); + out.flush(); + out.write(input, input.length / 2, input.length / 2); + out.finish(); + } + + try (CompressionInputStream in = + codec.createInputStream(new ByteArrayInputStream(compressed.toByteArray()))) { + assertThat(in.readAllBytes()).isEqualTo(input); + } + } + + @Test + public void testPositionAndReset() 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.resetState(); + assertThat(in.getPos()).isZero(); + assertThat(in.readAllBytes()).isEqualTo(input); + } + } + + 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 ByteArrayInputStream compress(BrotliCodec codec, byte[] input) throws IOException { + ByteArrayOutputStream compressed = new ByteArrayOutputStream(); + try (CompressionOutputStream out = codec.createOutputStream(compressed)) { + out.write(input); + out.finish(); + } + return new ByteArrayInputStream(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; + + @Override + public void close() throws IOException { + closed = true; + super.close(); + } + } } From d9f8a1d96b19c42724e8421f5533c818a46ad1d4 Mon Sep 17 00:00:00 2001 From: Donghun Ha Date: Tue, 27 Oct 2026 14:05:21 +0900 Subject: [PATCH 3/6] Bound Brotli decompression and fix stream contracts --- .../apache/parquet/hadoop/CodecFactory.java | 9 ++- .../parquet/hadoop/codec/BrotliCodec.java | 27 ++++--- .../codec/BrotliCompressionInputStream.java | 17 ++-- .../codec/BrotliCompressionOutputStream.java | 12 ++- .../parquet/hadoop/codec/TestBrotliCodec.java | 77 ++++++++++++++++--- 5 files changed, 106 insertions(+), 36 deletions(-) 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/codec/BrotliCodec.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/codec/BrotliCodec.java index e146e566b0..7af9fb3b19 100644 --- 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 @@ -42,6 +42,7 @@ 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; @@ -50,12 +51,7 @@ public class BrotliCodec implements CompressionCodec, Configurable { public void setConf(Configuration conf) { this.conf = conf; if (conf != null) { - int configuredQuality = conf.getInt(BROTLI_QUALITY, DEFAULT_QUALITY); - if (configuredQuality < 0 || configuredQuality > 11) { - throw new IllegalArgumentException( - "Invalid Brotli compression quality: " + configuredQuality + ". Expected 0..11."); - } - quality = configuredQuality; + quality = conf.getInt(BROTLI_QUALITY, DEFAULT_QUALITY); } } @@ -66,6 +62,9 @@ public Configuration getConf() { @Override public CompressionOutputStream createOutputStream(OutputStream out) throws IOException { + if (quality < 0 || quality > 11) { + throw new IllegalArgumentException("Invalid Brotli compression quality: " + quality + ". Expected 0..11."); + } ensureBrotliAvailable(); try { return new BrotliCompressionOutputStream(out, quality); @@ -75,8 +74,7 @@ public CompressionOutputStream createOutputStream(OutputStream out) throws IOExc } @Override - public CompressionOutputStream createOutputStream(OutputStream out, Compressor compressor) - throws IOException { + public CompressionOutputStream createOutputStream(OutputStream out, Compressor compressor) throws IOException { return createOutputStream(out); } @@ -92,17 +90,24 @@ public Compressor createCompressor() { @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. + */ + public CompressionInputStream createInputStream(InputStream in, int maxOutputSize) throws IOException { ensureBrotliAvailable(); try { - return new BrotliCompressionInputStream(in); + return new BrotliCompressionInputStream(in, maxOutputSize); } catch (LinkageError e) { throw missingBrotli(e); } } @Override - public CompressionInputStream createInputStream(InputStream in, Decompressor decompressor) - throws IOException { + public CompressionInputStream createInputStream(InputStream in, Decompressor decompressor) throws IOException { return createInputStream(in); } 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 index 72284ceddc..af60650a92 100644 --- 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 @@ -29,15 +29,20 @@ class BrotliCompressionInputStream extends CompressionInputStream { private final byte[] decompressed; private int position; - BrotliCompressionInputStream(InputStream in) throws IOException { + BrotliCompressionInputStream(InputStream in, int maxOutputSize) throws IOException { super(in); byte[] compressed = in.readAllBytes(); + if (compressed.length == 0) { + this.decompressed = new byte[0]; + return; + } + int outputLimit = maxOutputSize > 0 ? 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 - // one-shot decoder, which handles those streams correctly. - this.decompressed = Decoder.decompress(compressed, 0, compressed.length); - } catch (IOException e) { + // bounded one-shot decoder, which handles those streams without allowing unbounded output. + this.decompressed = Decoder.decompress(compressed, 0, compressed.length, outputLimit); + } catch (IOException | RuntimeException e) { throw new IOException("Failed to decompress Brotli stream", e); } } @@ -70,8 +75,8 @@ public int available() { } @Override - public void resetState() { - position = 0; + public void resetState() throws IOException { + throw new IOException("BrotliCompressionInputStream does not support resetState"); } @Override 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 index 411711bdd8..35939f43cd 100644 --- 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 @@ -35,8 +35,7 @@ class BrotliCompressionOutputStream extends CompressionOutputStream { BrotliCompressionOutputStream(OutputStream out, int quality) throws IOException { super(out); - this.encoder = - new BrotliOutputStream(new NonClosingOutputStream(out), Encoder.Parameters.create(quality)); + this.encoder = new BrotliOutputStream(new NonClosingOutputStream(out), Encoder.Parameters.create(quality)); } @Override @@ -54,11 +53,10 @@ public void write(byte[] b, int off, int len) throws IOException { @Override public void flush() throws IOException { ensureNoFailure(); - if (finished) { - out.flush(); - return; + if (!finished) { + encoder.flush(); } - encoder.flush(); + out.flush(); } @Override @@ -101,10 +99,10 @@ public void close() throws IOException { closeFailure.addSuppressed(e); } } + closed = true; if (closeFailure != null) { throw closeFailure; } - closed = true; } private void ensureWritable() throws IOException { 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 index c2df6d8d13..6a2b76e788 100644 --- 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 @@ -20,6 +20,7 @@ package org.apache.parquet.hadoop.codec; import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; import java.io.ByteArrayInputStream; import java.io.ByteArrayOutputStream; @@ -65,10 +66,10 @@ public void testLargeStreamThroughCodecFactory() throws Exception { byte[] input = largeCompressibleInput(); CodecFactory factory = new CodecFactory(new Configuration(false), 64 * 1024); try { - BytesInput compressed = factory.getCompressor(CompressionCodecName.BROTLI).compress( - BytesInput.from(input)); - BytesInput decompressed = factory.getDecompressor(CompressionCodecName.BROTLI).decompress( - compressed, input.length); + 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(); @@ -90,23 +91,53 @@ public void testFlushThenFinishRoundTrips() throws Exception { byte[] input = new byte[1024 * 1024]; new Random(11).nextBytes(input); BrotliCodec codec = newCodec(5); - ByteArrayOutputStream compressed = new ByteArrayOutputStream(); + TrackingOutputStream target = new TrackingOutputStream(); - try (CompressionOutputStream out = codec.createOutputStream(compressed)) { + try (CompressionOutputStream out = codec.createOutputStream(target)) { out.write(input, 0, input.length / 2); out.flush(); out.write(input, input.length / 2, input.length / 2); out.finish(); } - try (CompressionInputStream in = - codec.createInputStream(new ByteArrayInputStream(compressed.toByteArray()))) { + assertThat(target.flushes).isGreaterThan(0); + try (CompressionInputStream in = codec.createInputStream(new ByteArrayInputStream(target.toByteArray()))) { assertThat(in.readAllBytes()).isEqualTo(input); } } @Test - public void testPositionAndReset() throws Exception { + 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); + + CodecFactory factory = new CodecFactory(new Configuration(false), 64 * 1024); + try { + assertThatThrownBy(() -> factory.getDecompressor(CompressionCodecName.BROTLI) + .decompress(BytesInput.from(compressed), 1024)) + .isInstanceOf(IOException.class); + } finally { + factory.release(); + } + } + + @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.fail = true; + + assertThatThrownBy(out::finish).isInstanceOf(IOException.class); + assertThatThrownBy(out::finish).isInstanceOf(IOException.class); + } + + @Test + public void testPositionAndSeek() throws Exception { byte[] input = new byte[100]; new Random(7).nextBytes(input); BrotliCodec codec = newCodec(5); @@ -115,9 +146,10 @@ public void testPositionAndReset() throws Exception { assertThat(in.read()).isEqualTo(input[0] & 0xFF); assertThat(in.getPos()).isEqualTo(1); - in.resetState(); + in.seek(0); assertThat(in.getPos()).isZero(); assertThat(in.readAllBytes()).isEqualTo(input); + assertThatThrownBy(in::resetState).isInstanceOf(IOException.class); } } @@ -130,12 +162,16 @@ private static BrotliCodec newCodec(int quality) { } 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 new ByteArrayInputStream(compressed.toByteArray()); + return compressed.toByteArray(); } private static byte[] largeCompressibleInput() { @@ -149,11 +185,30 @@ private static byte[] largeCompressibleInput() { 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 ByteArrayOutputStream { + private boolean fail; + + @Override + public void flush() throws IOException { + if (fail) { + throw new IOException("flush failed"); + } + super.flush(); + } } } From 35e207965b4350fe90f9e0dcdbb1d24154603ba0 Mon Sep 17 00:00:00 2001 From: Donghun Ha Date: Tue, 27 Oct 2026 14:16:15 +0900 Subject: [PATCH 4/6] Add Brotli codec regression tests for bounds and stream contracts --- .../parquet/hadoop/codec/BrotliCodec.java | 8 ++- .../codec/BrotliCompressionOutputStream.java | 25 ++++++-- .../parquet/hadoop/codec/TestBrotliCodec.java | 60 ++++++++++++++++++- 3 files changed, 84 insertions(+), 9 deletions(-) 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 index 7af9fb3b19..ae7c6e7018 100644 --- 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 @@ -30,6 +30,7 @@ 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. @@ -63,7 +64,7 @@ public Configuration getConf() { @Override public CompressionOutputStream createOutputStream(OutputStream out) throws IOException { if (quality < 0 || quality > 11) { - throw new IllegalArgumentException("Invalid Brotli compression quality: " + quality + ". Expected 0..11."); + throw new BadConfigurationException("Invalid Brotli compression quality: " + quality + ". Expected 0..11."); } ensureBrotliAvailable(); try { @@ -88,6 +89,11 @@ 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); 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 index 35939f43cd..16c80fe885 100644 --- 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 @@ -41,22 +41,37 @@ class BrotliCompressionOutputStream extends CompressionOutputStream { @Override public void write(int b) throws IOException { ensureWritable(); - encoder.write(b); + try { + encoder.write(b); + } catch (IOException e) { + failure = e; + throw e; + } } @Override public void write(byte[] b, int off, int len) throws IOException { ensureWritable(); - encoder.write(b, off, len); + try { + encoder.write(b, off, len); + } catch (IOException e) { + failure = e; + throw e; + } } @Override public void flush() throws IOException { ensureNoFailure(); - if (!finished) { - encoder.flush(); + try { + if (!finished) { + encoder.flush(); + } + out.flush(); + } catch (IOException e) { + failure = e; + throw e; } - out.flush(); } @Override 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 index 6a2b76e788..26c720e946 100644 --- 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 @@ -31,6 +31,7 @@ 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; @@ -96,11 +97,13 @@ public void testFlushThenFinishRoundTrips() throws Exception { 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(); } - assertThat(target.flushes).isGreaterThan(0); try (CompressionInputStream in = codec.createInputStream(new ByteArrayInputStream(target.toByteArray()))) { assertThat(in.readAllBytes()).isEqualTo(input); } @@ -112,18 +115,62 @@ public void testBoundedDecompressionRejectsBomb() throws Exception { byte[] compressed = compressBytes(newCodec(5), zeros); assertThatThrownBy(() -> newCodec(5).createInputStream(new ByteArrayInputStream(compressed), 1024)) - .isInstanceOf(IOException.class); + .isInstanceOf(IOException.class) + .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); + .isInstanceOf(IOException.class) + .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); + } + + @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(); + } + + byte[] nonEmpty = new byte[1024]; + byte[] nonEmptyCompressed = compressBytes(newCodec(5), nonEmpty); + assertThatThrownBy(() -> newCodec(5).createInputStream(new ByteArrayInputStream(nonEmptyCompressed), -1)) + .isInstanceOf(IOException.class); + } + + @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.fail = true; + + assertThatThrownBy(out::finish).isInstanceOf(IOException.class); + assertThatThrownBy(out::close).isInstanceOf(IOException.class); + out.close(); + assertThat(target.closes).isEqualTo(1); + } + @Test public void testFinishFailureIsReportedOnRetry() throws Exception { FailingOutputStream target = new FailingOutputStream(); @@ -202,6 +249,7 @@ public void flush() throws IOException { private static final class FailingOutputStream extends ByteArrayOutputStream { private boolean fail; + private int closes; @Override public void flush() throws IOException { @@ -210,5 +258,11 @@ public void flush() throws IOException { } super.flush(); } + + @Override + public void close() throws IOException { + closes++; + super.close(); + } } } From 93054b410091165db03e1ad298efeb6250001ef9 Mon Sep 17 00:00:00 2001 From: Donghun Ha Date: Thu, 29 Oct 2026 23:29:47 +0900 Subject: [PATCH 5/6] fix: preserve legacy Brotli codec class alias Accept the legacy Hadoop Brotli codec name while keeping the brotli4j-backed codec as the implementation. Add regression coverage for legacy and current names, other codecs, and unknown or null classes. Based on the patch suggested by Fokko in apache/parquet-java#3880. --- .../hadoop/metadata/CompressionCodecName.java | 18 ++++- .../metadata/TestCompressionCodecName.java | 65 +++++++++++++++++++ 2 files changed, 81 insertions(+), 2 deletions(-) create mode 100644 parquet-common/src/test/java/org/apache/parquet/hadoop/metadata/TestCompressionCodecName.java diff --git a/parquet-common/src/main/java/org/apache/parquet/hadoop/metadata/CompressionCodecName.java b/parquet-common/src/main/java/org/apache/parquet/hadoop/metadata/CompressionCodecName.java index 5354e6c690..176328a382 100644 --- a/parquet-common/src/main/java/org/apache/parquet/hadoop/metadata/CompressionCodecName.java +++ b/parquet-common/src/main/java/org/apache/parquet/hadoop/metadata/CompressionCodecName.java @@ -39,17 +39,31 @@ public static CompressionCodecName fromConf(String name) { return valueOf(name.toUpperCase(Locale.ENGLISH)); } + // Brotli codec class from the removed jbrotli-based brotli-codec artifact. Still accepted so that + // jobs configured through mapred.output.compression.codec keep writing Brotli. + private static final String LEGACY_BROTLI_CODEC_CLASS = "org.apache.hadoop.io.compress.BrotliCodec"; + public static CompressionCodecName fromCompressionCodec(Class clazz) { if (clazz == null) { return UNCOMPRESSED; } - String name = clazz.getName(); + CompressionCodecName codec = fromCompressionCodecClassName(clazz.getName()); + if (codec == null) { + throw new CompressionCodecNotSupportedException(clazz); + } + return codec; + } + + static CompressionCodecName fromCompressionCodecClassName(String name) { for (CompressionCodecName codec : CompressionCodecName.values()) { if (name.equals(codec.getHadoopCompressionCodecClassName())) { return codec; } } - throw new CompressionCodecNotSupportedException(clazz); + if (LEGACY_BROTLI_CODEC_CLASS.equals(name)) { + return BROTLI; + } + return null; } public static CompressionCodecName fromParquet(CompressionCodec codec) { diff --git a/parquet-common/src/test/java/org/apache/parquet/hadoop/metadata/TestCompressionCodecName.java b/parquet-common/src/test/java/org/apache/parquet/hadoop/metadata/TestCompressionCodecName.java new file mode 100644 index 0000000000..43b8b4681c --- /dev/null +++ b/parquet-common/src/test/java/org/apache/parquet/hadoop/metadata/TestCompressionCodecName.java @@ -0,0 +1,65 @@ +/* + * 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.metadata; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import org.apache.parquet.hadoop.codec.CompressionCodecNotSupportedException; +import org.junit.jupiter.api.Test; + +public class TestCompressionCodecName { + + @Test + public void testLegacyBrotliCodecClassMapsToBrotli() { + assertThat(CompressionCodecName.fromCompressionCodecClassName("org.apache.hadoop.io.compress.BrotliCodec")) + .isEqualTo(CompressionCodecName.BROTLI); + } + + @Test + public void testBrotliCodecClassMapsToBrotli() { + assertThat(CompressionCodecName.fromCompressionCodecClassName("org.apache.parquet.hadoop.codec.BrotliCodec")) + .isEqualTo(CompressionCodecName.BROTLI); + } + + @Test + public void testCodecClassNamesRoundTrip() { + for (CompressionCodecName codec : CompressionCodecName.values()) { + if (codec.getHadoopCompressionCodecClassName() != null) { + assertThat(CompressionCodecName.fromCompressionCodecClassName( + codec.getHadoopCompressionCodecClassName())) + .isEqualTo(codec); + } + } + } + + @Test + public void testUnknownCodecClass() { + assertThat(CompressionCodecName.fromCompressionCodecClassName(String.class.getName())) + .isNull(); + assertThatThrownBy(() -> CompressionCodecName.fromCompressionCodec(String.class)) + .isInstanceOf(CompressionCodecNotSupportedException.class) + .hasMessage("codec not supported: java.lang.String"); + } + + @Test + public void testNullCodecClassIsUncompressed() { + assertThat(CompressionCodecName.fromCompressionCodec(null)).isEqualTo(CompressionCodecName.UNCOMPRESSED); + } +} From 14ba766761b6179e958dbb829499ca76b4d19a48 Mon Sep 17 00:00:00 2001 From: Donghun Ha Date: Fri, 30 Oct 2026 01:29:05 +0900 Subject: [PATCH 6/6] fix: close failed Brotli streams and enforce output bounds Release the native encoder even after write or flush failures, retaining the original error and any cleanup errors without try-with-resources self-suppression. Reject negative decode limits and non-empty output under a zero limit. Add six regression tests and strengthen existing exception assertions. --- .../parquet/hadoop/codec/BrotliCodec.java | 2 + .../codec/BrotliCompressionInputStream.java | 14 +- .../codec/BrotliCompressionOutputStream.java | 23 ++- .../parquet/hadoop/codec/TestBrotliCodec.java | 175 ++++++++++++++++-- 4 files changed, 189 insertions(+), 25 deletions(-) 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 index ae7c6e7018..d9d05960f7 100644 --- 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 @@ -102,6 +102,8 @@ public CompressionInputStream createInputStream(InputStream in) throws IOExcepti /** * 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(); 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 index af60650a92..9cfa50a01e 100644 --- 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 @@ -31,17 +31,27 @@ class BrotliCompressionInputStream extends CompressionInputStream { 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; } - int outputLimit = maxOutputSize > 0 ? maxOutputSize : 1; + // 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. - this.decompressed = Decoder.decompress(compressed, 0, compressed.length, outputLimit); + 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); } 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 index 16c80fe885..8783b77ff6 100644 --- 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 @@ -77,6 +77,10 @@ public void flush() throws IOException { @Override public void finish() throws IOException { ensureNoFailure(); + finishEncoder(); + } + + private void finishEncoder() throws IOException { if (finished) { return; } @@ -84,7 +88,9 @@ public void finish() throws IOException { encoder.close(); finished = true; } catch (IOException e) { - failure = e; + if (failure == null) { + failure = e; + } throw e; } } @@ -99,18 +105,25 @@ public void close() throws IOException { if (closed) { return; } - IOException closeFailure = null; + // 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 { - finish(); + // Release native resources even when a previous write or flush has failed. + finishEncoder(); } catch (IOException e) { - closeFailure = 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 { + } else if (closeFailure != e && closeFailure.getCause() != e) { closeFailure.addSuppressed(e); } } 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 index 26c720e946..e7d2375c97 100644 --- 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 @@ -22,9 +22,12 @@ 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; @@ -116,6 +119,7 @@ public void testBoundedDecompressionRejectsBomb() throws Exception { 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); @@ -123,6 +127,7 @@ public void testBoundedDecompressionRejectsBomb() throws Exception { 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(); @@ -141,7 +146,8 @@ public void testInvalidQualityOnlyRejectsCompression() throws Exception { } assertThatThrownBy(() -> codec.createOutputStream(new ByteArrayOutputStream())) - .isInstanceOf(BadConfigurationException.class); + .isInstanceOf(BadConfigurationException.class) + .hasMessage("Invalid Brotli compression quality: 99. Expected 0..11."); } @Test @@ -150,11 +156,111 @@ public void testEmptyAndZeroSizeStreams() throws Exception { 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; - byte[] nonEmpty = new byte[1024]; - byte[] nonEmptyCompressed = compressBytes(newCodec(5), nonEmpty); - assertThatThrownBy(() -> newCodec(5).createInputStream(new ByteArrayInputStream(nonEmptyCompressed), -1)) - .isInstanceOf(IOException.class); + 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 @@ -163,10 +269,10 @@ public void testCloseAfterFinishFailureIsIdempotent() throws Exception { BrotliCodec codec = newCodec(5); CompressionOutputStream out = codec.createOutputStream(target); out.write(new byte[100 * 1024]); - target.fail = true; + target.failOnFlush = true; - assertThatThrownBy(out::finish).isInstanceOf(IOException.class); - assertThatThrownBy(out::close).isInstanceOf(IOException.class); + assertThatThrownBy(out::finish).isInstanceOf(IOException.class).hasMessage("flush failed"); + assertThatThrownBy(out::close).isInstanceOf(IOException.class).hasMessageContaining("failed"); out.close(); assertThat(target.closes).isEqualTo(1); } @@ -177,10 +283,14 @@ public void testFinishFailureIsReportedOnRetry() throws Exception { BrotliCodec codec = newCodec(5); CompressionOutputStream out = codec.createOutputStream(target); out.write(new byte[100 * 1024]); - target.fail = true; + target.failOnFlush = true; - assertThatThrownBy(out::finish).isInstanceOf(IOException.class); - assertThatThrownBy(out::finish).isInstanceOf(IOException.class); + 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 @@ -196,7 +306,9 @@ public void testPositionAndSeek() throws Exception { in.seek(0); assertThat(in.getPos()).isZero(); assertThat(in.readAllBytes()).isEqualTo(input); - assertThatThrownBy(in::resetState).isInstanceOf(IOException.class); + assertThatThrownBy(in::resetState) + .isInstanceOf(IOException.class) + .hasMessage("BrotliCompressionInputStream does not support resetState"); } } @@ -208,6 +320,13 @@ private static BrotliCodec newCodec(int quality) { 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)); } @@ -247,22 +366,42 @@ public void flush() throws IOException { } } - private static final class FailingOutputStream extends ByteArrayOutputStream { - private boolean fail; + 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 (fail) { - throw new IOException("flush failed"); + if (failOnFlush) { + throw flushFailure; } - super.flush(); } @Override public void close() throws IOException { closes++; - super.close(); + if (failOnClose) { + throw closeFailure; + } } } }