MINOR: Update zstd-jni to 1.4.8-2 (apache#9957)

* The latest version zstd-jni doesn't use `RecyclingBufferPool` by default, so we
pass it via the relevant constructors to maintain the behavior before this
change.
* zstd-jni fixes an issue when using Alpine, see luben/zstd-jni#157.
* zstd 1.4.7 includes several months of improvements across many axis,
from performance to various fixes. Details: https://github.com/facebook/zstd/releases/tag/v1.4.7
* zstd 1.4.8 is a hotfix release, details: https://github.com/facebook/zstd/releases/tag/v1.4.8

Reviewers: Chia-Ping Tsai <[email protected]>

checkstyle/import-control.xml

@@ -152,6 +152,7 @@
       <allow pkg="org.apache.kafka.common.protocol" />
       <allow pkg="org.apache.kafka.common.protocol.types" />
       <allow pkg="org.apache.kafka.common.errors" />
+      <allow pkg="com.github.luben.zstd" />
     </subpackage>
 
     <subpackage name="header">

clients/src/main/java/org/apache/kafka/common/record/CompressionType.java

@@ -16,6 +16,8 @@
  */
 package org.apache.kafka.common.record;
 
+import com.github.luben.zstd.BufferPool;
+import com.github.luben.zstd.RecyclingBufferPool;
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.utils.ByteBufferInputStream;
 import org.apache.kafka.common.utils.ByteBufferOutputStream;
@@ -120,8 +122,10 @@ public InputStream wrapForInput(ByteBuffer inputBuffer, byte messageVersion, Buf
         public OutputStream wrapForOutput(ByteBufferOutputStream buffer, byte messageVersion) {
             try {
                 // Set input buffer (uncompressed) to 16 KB (none by default) to ensure reasonable performance
-                // in cases where the caller passes a small number of bytes to write (potentially a single byte)
-                return new BufferedOutputStream((OutputStream) ZstdConstructors.OUTPUT.invoke(buffer), 16 * 1024);
+                // in cases where the caller passes a small number of bytes to write (potentially a single byte).
+                // It's ok to reference `RecyclingBufferPool` since it doesn't load any native libraries
+                return new BufferedOutputStream((OutputStream) ZstdConstructors.OUTPUT.invoke(buffer, RecyclingBufferPool.INSTANCE),
+                    16 * 1024);
             } catch (Throwable e) {
                 throw new KafkaException(e);
             }
@@ -131,8 +135,10 @@ public OutputStream wrapForOutput(ByteBufferOutputStream buffer, byte messageVer
         public InputStream wrapForInput(ByteBuffer buffer, byte messageVersion, BufferSupplier decompressionBufferSupplier) {
             try {
                 // Set output buffer (uncompressed) to 16 KB (none by default) to ensure reasonable performance
-                // in cases where the caller reads a small number of bytes (potentially a single byte)
-                return new BufferedInputStream((InputStream) ZstdConstructors.INPUT.invoke(new ByteBufferInputStream(buffer)), 16 * 1024);
+                // in cases where the caller reads a small number of bytes (potentially a single byte).
+                // It's ok to reference `RecyclingBufferPool` since it doesn't load any native libraries.
+                return new BufferedInputStream((InputStream) ZstdConstructors.INPUT.invoke(new ByteBufferInputStream(buffer),
+                    RecyclingBufferPool.INSTANCE), 16 * 1024);
             } catch (Throwable e) {
                 throw new KafkaException(e);
             }
@@ -219,10 +225,11 @@ private static class SnappyConstructors {
     }
 
     private static class ZstdConstructors {
+        // It's ok to reference `BufferPool` since it doesn't load any native libraries
         static final MethodHandle INPUT = findConstructor("com.github.luben.zstd.ZstdInputStream",
-            MethodType.methodType(void.class, InputStream.class));
+            MethodType.methodType(void.class, InputStream.class, BufferPool.class));
         static final MethodHandle OUTPUT = findConstructor("com.github.luben.zstd.ZstdOutputStream",
-            MethodType.methodType(void.class, OutputStream.class));
+            MethodType.methodType(void.class, OutputStream.class, BufferPool.class));
     }
 
     private static MethodHandle findConstructor(String className, MethodType methodType) {

gradle/dependencies.gradle

@@ -118,7 +118,7 @@ versions += [
   testRetryPlugin: "1.2.0",
   zinc: "1.3.5",
   zookeeper: "3.5.8",
-  zstd: "1.4.5-12"
+  zstd: "1.4.8-2"
 ]
 libs += [
   activation: "javax.activation:activation:$versions.activation",