Update zstd-jni to 1.5.7-25 and call setLongMax directly

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
 

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"

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)
     }

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)
 

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