You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by do...@apache.org on 2026/10/22 12:35:27 UTC
(spark) branch branch-4.0 updated: [SPARK-60432][CORE] Validate `spark.io.compression.zstd.bufferSize`
This is an automated email from the ASF dual-hosted git repository.
dongjoon-hyun pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new c68e5d919345 [SPARK-60432][CORE] Validate `spark.io.compression.zstd.bufferSize`
c68e5d919345 is described below
commit c68e5d9193457f3f572e1cedd4aec934d8a37f5d
Author: Dongjoon Hyun <do...@apache.org>
AuthorDate: Thu Oct 22 05:19:05 2026 -0700
[SPARK-60432][CORE] Validate `spark.io.compression.zstd.bufferSize`
### What changes were proposed in this pull request?
This PR aims to validate `spark.io.compression.zstd.bufferSize` via `checkValue` and to document its valid range in `docs/configuration.md`. Like the existing `spark.shuffle.file.buffer` validation, the value must be positive and less than or equal to `ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH` (`2147483632`) bytes.
Note that this PR doesn't validate `spark.io.compression.zstd.strategy` because zstd may add new strategies in the future. This is consistent with the decision in [SPARK-52924](https://github.com/apache/spark/pull/51630#discussion_r2225184049).
### Why are the changes needed?
Currently, `spark.io.compression.zstd.bufferSize` accepts any value. `ZStdCompressionCodec` reads it with `.toInt`, so `0`, `-1`, `2g` (overflows to `-2147483648`), and `4g` (overflows to `0`) all become invalid buffer sizes.
An invalid value is not detected until `ZStdCompressionCodec` wraps a stream. At that point, the zstd stream has already allocated its native context, and neither `ZstdOutputStreamNoFinalizer` nor `ZstdInputStreamNoFinalizer` has a finalizer or a `Cleaner`. So, each failed wrap leaves an unreachable zstd stream whose native context is never freed. This happens for shuffle, broadcast, event logs, and the standalone Master recovery store.
https://github.com/apache/spark/blob/cd06735c008285bddb9bd976b3e931f37eb4990f/core/src/main/scala/org/apache/spark/io/CompressionCodec.scala#L240-L262
```scala
// The native context is allocated here.
val os = new ZstdOutputStreamNoFinalizer(s, bufferPool).setLevel(level).setWorkers(workers)
strategy.foreach(os.setStrategy)
// IllegalArgumentException: Buffer size <= 0
new BufferedOutputStream(os, bufferSize)
```
With this PR, Spark fails to create `ZStdCompressionCodec` before allocating any zstd stream.
Note that `spark.io.compression.zstd.bufferSize` has been available since [v2.3.0](https://github.com/apache/spark/releases/tag/v2.3.0) (2018-02-22).
### Does this PR introduce _any_ user-facing change?
Yes, but only for invalid values. Previously, they were accepted, but every zstd stream creation failed after leaking a native zstd context. Now, Spark rejects them when it creates `ZStdCompressionCodec`.
- **Before**
```
java.lang.IllegalArgumentException: Buffer size <= 0
```
- **After**
```
org.apache.spark.SparkIllegalArgumentException: [INVALID_CONF_VALUE.REQUIREMENT] The value '0b' in the config "spark.io.compression.zstd.bufferSize" is invalid. The buffer size must be positive and less than or equal to 2147483632. SQLSTATE: 22022
```
### How was this patch tested?
Pass the CIs with the newly added test case in `CompressionCodecSuite`. It checks that `1` and `2147483632` are accepted while `0`, `-1`, `2147483633`, `2g`, and `4g` are rejected.
I also verified that the new test case fails without the main code change.
```
$ build/sbt 'core/testOnly *CompressionCodecSuite'
...
[info] Tests: succeeded 16, failed 0, canceled 0, ignored 0, pending 0
[info] All tests passed.
```
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
Closes #59679 from dongjoon-hyun/SPARK-60432.
Authored-by: Dongjoon Hyun <do...@apache.org>
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 18eb701a85be5433be59c9c3734a681faee6ae55)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
(cherry picked from commit 25739585554ba9a9f0a82919df07a3cd6ecb75d5)
Signed-off-by: Dongjoon Hyun <do...@apache.org>
---
.../org/apache/spark/internal/config/package.scala | 3 +++
.../apache/spark/io/CompressionCodecSuite.scala | 27 +++++++++++++++++++++-
docs/configuration.md | 1 +
3 files changed, 30 insertions(+), 1 deletion(-)
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index 506a132e1093..7dc161310e26 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -2071,6 +2071,9 @@ package object config {
"cost because of excessive JNI call overhead")
.version("2.3.0")
.bytesConf(ByteUnit.BYTE)
+ .checkValue(v => v > 0 && v <= ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH,
+ s"The buffer size must be positive and less than or equal to" +
+ s" ${ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH}.")
.createWithDefaultString("32k")
private[spark] val IO_COMPRESSION_ZSTD_BUFFERPOOL_ENABLED =
diff --git a/core/src/test/scala/org/apache/spark/io/CompressionCodecSuite.scala b/core/src/test/scala/org/apache/spark/io/CompressionCodecSuite.scala
index ff971b72d891..25295a097fb6 100644
--- a/core/src/test/scala/org/apache/spark/io/CompressionCodecSuite.scala
+++ b/core/src/test/scala/org/apache/spark/io/CompressionCodecSuite.scala
@@ -23,7 +23,8 @@ import java.util.Locale
import com.google.common.io.ByteStreams
import org.apache.spark.{SparkConf, SparkFunSuite, SparkIllegalArgumentException}
-import org.apache.spark.internal.config.IO_COMPRESSION_ZSTD_BUFFERPOOL_ENABLED
+import org.apache.spark.internal.config.{IO_COMPRESSION_ZSTD_BUFFERPOOL_ENABLED, IO_COMPRESSION_ZSTD_BUFFERSIZE}
+import org.apache.spark.unsafe.array.ByteArrayMethods
class CompressionCodecSuite extends SparkFunSuite {
val conf = new SparkConf(false)
@@ -127,6 +128,30 @@ class CompressionCodecSuite extends SparkFunSuite {
testConcatenationOfSerializedStreams(codec)
}
+ test("SPARK-60432: zstd compression codec validates the buffer size") {
+ val maxBufferSize = ByteArrayMethods.MAX_ROUNDED_ARRAY_LENGTH
+ testCodec(new ZStdCompressionCodec(
+ new SparkConf(false).set(IO_COMPRESSION_ZSTD_BUFFERSIZE.key, "1")))
+ // The codec allocates the buffer per stream, so creating it with the upper bound is cheap.
+ new ZStdCompressionCodec(
+ new SparkConf(false).set(IO_COMPRESSION_ZSTD_BUFFERSIZE.key, maxBufferSize.toString))
+
+ Seq(
+ "0" -> "0",
+ "-1" -> "-1",
+ s"${maxBufferSize + 1}" -> s"${maxBufferSize + 1}",
+ "2g" -> "2147483648",
+ "4g" -> "4294967296").foreach { case (bufferSize, confValue) =>
+ val conf = new SparkConf(false).set(IO_COMPRESSION_ZSTD_BUFFERSIZE.key, bufferSize)
+ val e = intercept[IllegalArgumentException] {
+ new ZStdCompressionCodec(conf)
+ }
+ assert(e.getMessage ===
+ s"'$confValue' in ${IO_COMPRESSION_ZSTD_BUFFERSIZE.key} is invalid. " +
+ s"The buffer size must be positive and less than or equal to $maxBufferSize.")
+ }
+ }
+
test("bad compression codec") {
checkError(
exception = intercept[SparkIllegalArgumentException] {
diff --git a/docs/configuration.md b/docs/configuration.md
index 5a14af714d13..4eade0c51414 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -1923,6 +1923,7 @@ Apart from these, the following properties are also available, and may be useful
is used. Lowering this size will lower the shuffle memory usage when Zstd is used, but it
might increase the compression cost because of excessive JNI call overhead. This
configuration only applies to <code>spark.io.compression.codec</code>.
+ The value must be positive and less than or equal to 2147483632 bytes.
</td>
<td>2.3.0</td>
</tr>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org