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:23:32 UTC
(spark) branch branch-4.1 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.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new e4db4fe30705 [SPARK-60432][CORE] Validate `spark.io.compression.zstd.bufferSize`
e4db4fe30705 is described below
commit e4db4fe30705514dd939f490888fd28511bcafac
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>
---
.../org/apache/spark/internal/config/package.scala | 3 +++
.../apache/spark/io/CompressionCodecSuite.scala | 31 +++++++++++++++++++++-
docs/configuration.md | 1 +
3 files changed, 34 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 31871c2ac2b8..efcd4a7635f6 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
@@ -2109,6 +2109,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 d6f0bfd237e4..92b1741c1e8c 100644
--- a/core/src/test/scala/org/apache/spark/io/CompressionCodecSuite.scala
+++ b/core/src/test/scala/org/apache/spark/io/CompressionCodecSuite.scala
@@ -21,7 +21,8 @@ import java.io.{ByteArrayInputStream, ByteArrayOutputStream}
import java.util.Locale
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
import org.apache.spark.util.Utils
class CompressionCodecSuite extends SparkFunSuite {
@@ -126,6 +127,34 @@ 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" -> "0b",
+ "-1" -> "-1b",
+ s"${maxBufferSize + 1}" -> s"${maxBufferSize + 1}b",
+ "2g" -> "2147483648b",
+ "4g" -> "4294967296b").foreach { case (bufferSize, confValue) =>
+ val conf = new SparkConf(false).set(IO_COMPRESSION_ZSTD_BUFFERSIZE.key, bufferSize)
+ checkError(
+ exception = intercept[SparkIllegalArgumentException] {
+ new ZStdCompressionCodec(conf)
+ },
+ condition = "INVALID_CONF_VALUE.REQUIREMENT",
+ parameters = Map(
+ "confName" -> IO_COMPRESSION_ZSTD_BUFFERSIZE.key,
+ "confValue" -> confValue,
+ "confRequirement" ->
+ 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 229b494a3dea..a44d0a5708d9 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -1994,6 +1994,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