From bd4e18042dcddcd93b1cbe26c85f37af7f1b5fb7 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Wed, 28 Oct 2026 14:49:19 +0000 Subject: [PATCH] Update zstd-jni to 1.5.7-25 and call setLongMax directly Motivation: #3710 looked up ZstdDecompressCtx.setLongMax with a MethodHandle because no zstd-jni release had it. zstd-jni 1.5.7-25 includes it, so the lookup can go. Modification: - Update zstd-jni from 1.5.7-24 to 1.5.7-25. - Call ZstdDecompressCtx.setLongMax directly instead of through a MethodHandle. zstd-jni is an optional dependency, so a NoSuchMethodError from an older version is still turned into an UnsupportedOperationException. - Drop the test branch for a zstd-jni without setLongMax, and update the docs to name 1.5.7-25 as the minimum version for withWindowLogMax. Result: The zstd decompression window limit works with the zstd-jni version Pekko depends on. Tests: - sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.ZstdSpec org.apache.pekko.stream.io.compression.CompressionWithoutZstdSpec" (42 passed) - sbt "++3.3.8 stream/compile" - scalafmt on the changed Scala files References: Refs #3674, #3710 --- .../operators/Compression/zstdDecompress.md | 3 +- project/Dependencies.scala | 2 +- .../stream/io/compression/ZstdSpec.scala | 25 +++++--------- .../stream/ZstdDecompressionSettings.scala | 5 ++- .../io/compression/ZstdDecompressor.scala | 33 ++++++------------- 5 files changed, 22 insertions(+), 46 deletions(-) diff --git a/docs/src/main/paradox/stream/operators/Compression/zstdDecompress.md b/docs/src/main/paradox/stream/operators/Compression/zstdDecompress.md index 5750aef734..b08e67ee26 100644 --- a/docs/src/main/paradox/stream/operators/Compression/zstdDecompress.md +++ b/docs/src/main/paradox/stream/operators/Compression/zstdDecompress.md @@ -34,8 +34,7 @@ By default, zstd accepts frames with a window of up to 128 MiB (a window log of the window buffer from the frame header before any data is decompressed. When decompressing untrusted input, use `withWindowLogMax` to reject frames with a larger window, for example a window log of 23 (8 MiB), the limit that [RFC 9659](https://www.rfc-editor.org/rfc/rfc9659) sets for the HTTP `zstd` content-coding. `withWindowLogMax` -requires a zstd-jni version newer than 1.5.7-24, which provides `ZstdDecompressCtx.setLongMax`; with an older -version, the operator fails when it is materialized. +requires zstd-jni 1.5.7-25 or later; with an older version, the operator fails when it is materialized. ## Reactive Streams semantics diff --git a/project/Dependencies.scala b/project/Dependencies.scala index f31ce9896d..6f3c83093a 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -49,7 +49,7 @@ object Dependencies { val reactiveStreamsVersion = "1.0.4" - val zstdJniVersion = "1.5.7-24" + val zstdJniVersion = "1.5.7-25" val scalaTestVersion = "3.2.20" val scalaTestScalaCheckVersion = "1-19" diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdSpec.scala index 4717ca4175..ce9dc4790f 100644 --- a/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdSpec.scala +++ b/stream-tests/src/test/scala/org/apache/pekko/stream/io/compression/ZstdSpec.scala @@ -27,7 +27,6 @@ import scala.util.Random import com.github.luben.zstd.{ Zstd, ZstdCompressCtx, - ZstdDecompressCtx, ZstdException, ZstdInputStream, ZstdInputStreamNoFinalizer, @@ -146,22 +145,14 @@ class ZstdSpec extends CoderSpec[ZstdException]("zstd") { .join .awaitResult(3.seconds) - // ZstdDecompressCtx.setLongMax is newer than zstd-jni 1.5.7-24 (#3674) - val supported = classOf[ZstdDecompressCtx].getMethods.exists(_.getName == "setLongMax") - if (supported) { - val ex = the[RuntimeException] thrownBy decompress(23) - ex.ultimateCause shouldBe a[ZstdException] - decompress(24) should ===(input) - Source - .single(compressed) - .via(JavaCompression.zstdDecompress(ZstdDecompressionSettings.create().withWindowLogMax(24))) - .join - .awaitResult(3.seconds) should ===(input) - } else { - // a configured limit must not be silently ignored - val ex = the[Exception] thrownBy decompress(24) - ex.ultimateCause shouldBe an[UnsupportedOperationException] - } + val ex = the[RuntimeException] thrownBy decompress(23) + ex.ultimateCause shouldBe a[ZstdException] + decompress(24) should ===(input) + Source + .single(compressed) + .via(JavaCompression.zstdDecompress(ZstdDecompressionSettings.create().withWindowLogMax(24))) + .join + .awaitResult(3.seconds) should ===(input) // without a limit, the libzstd default of 2^27 applies ourDecode(compressed) should ===(input) } diff --git a/stream/src/main/scala/org/apache/pekko/stream/ZstdDecompressionSettings.scala b/stream/src/main/scala/org/apache/pekko/stream/ZstdDecompressionSettings.scala index 5525b9446d..4047edb5eb 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/ZstdDecompressionSettings.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/ZstdDecompressionSettings.scala @@ -51,9 +51,8 @@ final class ZstdDecompressionSettings private ( * greater or equal to `Compression.ZstdMinWindowLog` and less than or equal to `Compression.ZstdMaxWindowLog`. * For example, RFC 9659 limits the HTTP `zstd` content-coding to a window log of 23 (8 MiB). * - * A value other than 0 requires a zstd-jni version that provides `ZstdDecompressCtx.setLongMax`, which is newer - * than 1.5.7-24. With an older version, materializing the decompression operator fails with an - * `UnsupportedOperationException`. + * A value other than 0 requires zstd-jni 1.5.7-25 or later. With an older version, materializing the + * decompression operator fails with an `UnsupportedOperationException`. */ def withWindowLogMax(windowLogMax: Int): ZstdDecompressionSettings = copy(windowLogMax = windowLogMax) diff --git a/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDecompressor.scala b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDecompressor.scala index c9c6091d50..3f47d9281d 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDecompressor.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/impl/io/compression/ZstdDecompressor.scala @@ -17,7 +17,6 @@ package org.apache.pekko.stream.impl.io.compression -import java.lang.invoke.{ MethodHandle, MethodHandles, MethodType } import java.nio.ByteBuffer import scala.annotation.tailrec @@ -37,28 +36,16 @@ import pekko.util.ByteString private val EmptyInput = ByteBuffer.allocate(0) private val NoInput: Iterator[ByteBuffer] = Iterator.empty - // ZstdDecompressCtx.setLongMax was added after zstd-jni 1.5.7-24 (luben/zstd-jni#431). zstd-jni is an optional - // dependency that users add themselves, so the method is looked up rather than linked against, and is used when a - // newer zstd-jni is on the classpath. - // TODO call setLongMax directly once Pekko depends on a zstd-jni release that has it (#3674) - private lazy val setLongMax: Option[MethodHandle] = - try - Some( - MethodHandles - .publicLookup() - .findVirtual( - classOf[ZstdDecompressCtx], - "setLongMax", - MethodType.methodType(classOf[ZstdDecompressCtx], Integer.TYPE)) - .asType(MethodType.methodType(classOf[AnyRef], classOf[AnyRef], Integer.TYPE))) - catch { case _: NoSuchMethodException => None } - - private def setWindowLogMax(ctx: ZstdDecompressCtx, windowLogMax: Int): Unit = { - val handle = setLongMax.getOrElse(throw new UnsupportedOperationException( - "ZstdDecompressionSettings.withWindowLogMax requires a zstd-jni version that provides " + - "ZstdDecompressCtx.setLongMax, which is newer than 1.5.7-24")) - val _: AnyRef = handle.invokeExact(ctx.asInstanceOf[AnyRef], windowLogMax) - } + // zstd-jni is an optional dependency that users add themselves, so an older version without + // ZstdDecompressCtx.setLongMax may be on the classpath + private def setWindowLogMax(ctx: ZstdDecompressCtx, windowLogMax: Int): Unit = + try ctx.setLongMax(windowLogMax) + catch { + case e: NoSuchMethodError => + throw new UnsupportedOperationException( + "ZstdDecompressionSettings.withWindowLogMax requires zstd-jni 1.5.7-25 or later", + e) + } // The level only applies to compression. Each materialization loads the dictionary into its own // context, which owns the digested dictionary and frees it on close. A shared ZstdDictDecompress