This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new cebc1cef49 Add zstd compression to the Jackson serializers (#3641)
cebc1cef49 is described below
commit cebc1cef49e0466e3fe5cc72ad465009b2b7fd51
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 23 08:07:47 2026 +0100
Add zstd compression to the Jackson serializers (#3641)
* Add zstd compression to the Jackson serializers
Motivation:
The Jackson serializers support gzip and lz4 payload compression. zstd
compresses nearly as well as gzip while being much faster.
Modification:
- Add `algorithm = zstd` and a `zstd-level` setting for the jackson and
jackson3 serializers, using zstd-jni as an optional dependency.
- Detect zstd payloads on deserialization by the zstd frame magic number,
whatever the configured algorithm.
- Apply max-decompressed-size to zstd payloads: a declared decompressed
size beyond the limit is rejected before allocating, the declared size
bounds the output, and frames without a declared size are decompressed
in chunks and rejected once past the limit.
- Report a missing zstd-jni as an IllegalArgumentException.
- Pre-size the buffers used for gzip and zstd (de)compression.
- Document zstd, its dependency and how to enable it in a running
cluster.
Result:
Jackson serialized payloads can be compressed with zstd.
Tests:
- sbt "serialization-jackson/test" "serialization-jackson3/test": passed
- sbt "serialization-jackson/mimaReportBinaryIssues": passed
- sbt "serialization-jackson/makePom" "serialization-jackson/packageBin":
zstd-jni is optional in the POM and the OSGi Import-Package
References:
Refs #3639
* Avoid extra copies and small buffers in Jackson lz4 and gzip compression
Motivation:
lz4 compression copied the compressed bytes twice (trimming, then
prepending the header) and the gzip streams used their default 512 byte
buffer.
Modification:
- Compress lz4 directly after the header in one array, which is only
copied once to trim it. The wire format is unchanged.
- Use 4 KiB buffers for GZIPOutputStream and GZIPInputStream.
Result:
Less copying and fewer small writes when compressing payloads.
Tests:
- sbt "serialization-jackson/test" "serialization-jackson3/test": passed,
including a check that lz4 payloads are unchanged byte for byte
- sbt "serialization-jackson/mimaReportBinaryIssues": passed
References:
Refs #3641
* Keep the forked test JVM alive in JacksonWithoutZstdSpec
Motivation:
CI's forked test JVM for serialization-jackson and serialization-jackson3
exited
with code 255. The spec closed its isolated class loader while dispatcher
shutdown tasks of the terminated ActorSystem were still loading classes
from it.
The NoClassDefFoundError was fatal, so pekko.jvm-exit-on-fatal-error stopped
the JVM.
Modification:
Do not close the isolated class loader, and set
pekko.jvm-exit-on-fatal-error = off for the nested ActorSystem, as
CompressionWithoutZstdSpec in stream-tests already does. Use
a[ClassNotFoundException] and StandardCharsets.UTF_8.
Result:
A late class load after termination can no longer exit the forked test JVM.
Tests:
sbt "serialization-jackson/test" "serialization-jackson3/test" (162 and 160
passed)
sbt "serialization-jackson/testOnly *JacksonWithoutZstdSpec"
"serialization-jackson3/testOnly *JacksonWithoutZstdSpec"
References:
https://github.com/apache/pekko/actions/runs/39802277990/job/119906715764?pr=3641
---
docs/src/main/paradox/serialization-jackson.md | 20 ++-
project/Dependencies.scala | 8 +
project/OSGi.scala | 8 +-
project/Paradox.scala | 1 +
.../src/main/resources/reference.conf | 9 +-
.../serialization/jackson/JacksonSerializer.scala | 67 ++++++--
.../serialization/jackson/ZstdCompression.scala | 106 +++++++++++++
.../jackson/JacksonSerializerBufferSizeSpec.scala | 49 ++++++
.../jackson/JacksonSerializerSpec.scala | 172 ++++++++++++++++++++-
.../jackson/JacksonWithoutZstdSpec.scala | 119 ++++++++++++++
.../src/main/resources/reference.conf | 9 +-
.../serialization/jackson3/JacksonSerializer.scala | 67 ++++++--
.../serialization/jackson3/ZstdCompression.scala | 106 +++++++++++++
.../jackson3/JacksonSerializerBufferSizeSpec.scala | 49 ++++++
.../jackson3/JacksonSerializerSpec.scala | 172 ++++++++++++++++++++-
.../jackson3/JacksonWithoutZstdSpec.scala | 119 ++++++++++++++
16 files changed, 1054 insertions(+), 27 deletions(-)
diff --git a/docs/src/main/paradox/serialization-jackson.md
b/docs/src/main/paradox/serialization-jackson.md
index b55ee6b189..0db4bfb3d6 100644
--- a/docs/src/main/paradox/serialization-jackson.md
+++ b/docs/src/main/paradox/serialization-jackson.md
@@ -482,10 +482,24 @@ the `jackson-json` binding the default configuration is:
@@snip
[reference.conf](/serialization-jackson/src/main/resources/reference.conf) {
#compression }
-Supported compression algorithms are: gzip, lz4. Use 'off' to disable
compression.
-Gzip is generally slower than lz4.
+Supported compression algorithms are: gzip, lz4, zstd. Use 'off' to disable
compression.
+Gzip is generally slower than lz4 and zstd. zstd usually compresses better
than lz4 and nearly as well as gzip,
+while being much faster than gzip. Its compression level can be set with the
`zstd-level` property (default 3).
Messages larger than the `compress-larger-than` property are compressed.
+zstd compression uses [zstd-jni](https://github.com/luben/zstd-jni), an
optional dependency that has to be
+added to use it:
+
+@@dependency[sbt,Maven,Gradle] {
+ group=com.github.luben
+ artifact=zstd-jni
+ version="$zstd_jni_version$"
+}
+
+Every node that may receive zstd compressed messages must be able to
decompress them, so when changing the
+`algorithm` to `zstd` in a running cluster, first deploy the zstd-jni
dependency (and a Pekko version that
+supports zstd) to all nodes with the old `algorithm`, and then change the
`algorithm` in a second rolling update.
+
Compression can be disabled by setting the `algorithm` property to `off`. It
will still be able to decompress
payloads that were compressed when serialized, e.g. if this configuration is
changed.
@@ -493,7 +507,7 @@ For the `jackson-cbor` and custom bindings other than
`jackson-json` compression
but can be enabled in the same way as the configuration shown above but
replacing `jackson-json` with
the binding name (for example `jackson-cbor`).
-The `compression.max-decompressed-size` property limits how large a compressed
(gzip or lz4) payload may
+The `compression.max-decompressed-size` property limits how large a compressed
(gzip, lz4 or zstd) payload may
become when it is decompressed during deserialization, guarding against a
small message that expands to an
excessive size. It applies regardless of the `algorithm` setting. The default
is `unlimited`, which applies
no limit (a negative number also means unlimited). Set a size such as `256
MiB`, larger than any payload the
diff --git a/project/Dependencies.scala b/project/Dependencies.scala
index 5b59e82145..533da66d57 100644
--- a/project/Dependencies.scala
+++ b/project/Dependencies.scala
@@ -49,6 +49,8 @@ object Dependencies {
val reactiveStreamsVersion = "1.0.4"
+ val zstdJniVersion = "1.5.7-24"
+
val scalaTestVersion = "3.2.20"
val scalaTestScalaCheckVersion = "1-19"
val scalaCheckVersion = "1.19.0"
@@ -108,6 +110,8 @@ object Dependencies {
val lz4Java = "at.yawk.lz4" % "lz4-java" % "1.12.1"
+ val zstdJni = "com.github.luben" % "zstd-jni" % zstdJniVersion
+
val logback = "ch.qos.logback" % "logback-classic" % logbackVersion
val jspecify = "org.jspecify" % "jspecify" % "1.0.1" % Optional
@@ -339,6 +343,8 @@ object Dependencies {
jacksonCbor,
jacksonScala,
lz4Java,
+ // optional, only needed for zstd compression
+ zstdJni % Optional,
TestDependencies.junit,
TestDependencies.scalatest)
@@ -349,6 +355,8 @@ object Dependencies {
jacksonCbor3,
jacksonScala3,
lz4Java,
+ // optional, only needed for zstd compression
+ zstdJni % Optional,
TestDependencies.junit,
TestDependencies.scalatest)
diff --git a/project/OSGi.scala b/project/OSGi.scala
index cbff8a47ef..d54563cb5b 100644
--- a/project/OSGi.scala
+++ b/project/OSGi.scala
@@ -88,8 +88,12 @@ object OSGi {
// (including typical ones such as scala runtime)
OsgiKeys.explodedJars := Seq(assembly.value))
- lazy val jackson = exports(Seq("org.apache.pekko.serialization.jackson.*"))
- lazy val jackson3 = exports(Seq("org.apache.pekko.serialization.jackson3.*"))
+ lazy val jackson = exports(
+ Seq("org.apache.pekko.serialization.jackson.*"),
+ imports = Seq(optionalResolution("com.github.luben.zstd.*")))
+ lazy val jackson3 = exports(
+ Seq("org.apache.pekko.serialization.jackson3.*"),
+ imports = Seq(optionalResolution("com.github.luben.zstd.*")))
lazy val remote = exports(Seq("org.apache.pekko.remote.*"))
diff --git a/project/Paradox.scala b/project/Paradox.scala
index ddf16efc0d..a89571b2ab 100644
--- a/project/Paradox.scala
+++ b/project/Paradox.scala
@@ -73,6 +73,7 @@ object Paradox {
"sigar_loader.version" -> "1.6.6-rev002",
"aeron_version" -> Dependencies.aeronVersion,
"netty_version" -> Dependencies.nettyVersion,
+ "zstd_jni_version" -> Dependencies.zstdJniVersion,
"logback_version" -> Dependencies.logbackVersion))
lazy val rootsSettings = Seq(
diff --git a/serialization-jackson/src/main/resources/reference.conf
b/serialization-jackson/src/main/resources/reference.conf
index 4de5217f79..7831fec131 100644
--- a/serialization-jackson/src/main/resources/reference.conf
+++ b/serialization-jackson/src/main/resources/reference.conf
@@ -200,13 +200,19 @@ pekko.serialization.jackson {
# Compression algorithm.
# - off : no compression
# - gzip : using common java gzip
+ # - lz4 : using lz4-java
+ # - zstd : using zstd-jni, an optional dependency that has to be added to
the classpath
algorithm = off
# If compression is enabled with the `algorithm` setting the payload is
compressed
# when it's larger than this value.
compress-larger-than = 0 KiB
- # Maximum size of a payload after decompression. A compressed (gzip or lz4)
+ # zstd compression level used when `algorithm` is zstd, from -131072
(fastest) to
+ # 22 (strongest compression). 3 is zstd's default level.
+ zstd-level = 3
+
+ # Maximum size of a payload after decompression. A compressed (gzip, lz4
or zstd)
# payload that decompresses to more than this is rejected rather than
# allocated, guarding against a small message that inflates without bound.
# This applies on deserialization regardless of the `algorithm` setting
above.
@@ -262,6 +268,7 @@ pekko.serialization.jackson.jackson-json.compression {
# - off : no compression
# - gzip : using common java gzip
# - lz4 : using lz4-java
+ # - zstd : using zstd-jni, an optional dependency that has to be added to
the classpath
algorithm = gzip
# If compression is enabled with the `algorithm` setting the payload is
compressed
diff --git
a/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala
b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala
index 1ef5094b60..a7c6815da7 100644
---
a/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala
+++
b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/JacksonSerializer.scala
@@ -139,6 +139,39 @@ import pekko.util.OptionVal
LZ4Meta.get(bytes).isDefined
}
+ private final val MinBufferSize = 4 * 1024
+ private final val MaxInitialDecompressBufferSize = 16 * 1024 * 1024
+
+ /**
+ * Initial size of the buffer that a compressed payload is decompressed into
when its
+ * decompressed size is not known up front. JSON and CBOR typically compress
3-10x, so 4x the
+ * compressed size avoids most resizing without allocating much more than
needed. It is never
+ * larger than the configured maximum decompressed size allows, and capped
so that a large
+ * payload grows the buffer rather than allocating a huge one up front.
+ */
+ def initialDecompressBufferSize(compressedSize: Int, maxDecompressedSize:
Long): Int = {
+ val estimate = compressedSize.toLong * 4
+ val bounded = if (maxDecompressedSize >= 0) math.min(estimate,
maxDecompressedSize + 1) else estimate
+ math.max(MinBufferSize, math.min(bounded,
MaxInitialDecompressBufferSize.toLong)).toInt
+ }
+
+ /**
+ * Initial size of the buffer that a payload is compressed into: a quarter
of its size (JSON and
+ * CBOR typically compress at least that well), but at least 4 KiB.
+ */
+ def initialCompressBufferSize(uncompressedSize: Int): Int =
+ math.max(MinBufferSize, uncompressedSize / 4)
+
+ // a zstd frame starts with the magic number 0xFD2FB528, stored little-endian
+ def isZstd(bytes: Array[Byte]): Boolean = {
+ (bytes ne null) &&
+ (bytes.length >= 4) &&
+ (bytes(0) == 0x28.toByte) &&
+ (bytes(1) == 0xB5.toByte) &&
+ (bytes(2) == 0x2F.toByte) &&
+ (bytes(3) == 0xFD.toByte)
+ }
+
}
/**
@@ -168,6 +201,7 @@ import pekko.util.OptionVal
object Off extends Algorithm
final case class GZip(largerThan: Long) extends Algorithm
final case class LZ4(largerThan: Long) extends Algorithm
+ final case class Zstd(largerThan: Long, level: Int) extends Algorithm
}
/**
@@ -202,10 +236,15 @@ import pekko.util.OptionVal
case "lz4" =>
val compressLargerThan =
conf.getBytes("compression.compress-larger-than")
Compression.LZ4(compressLargerThan)
+ case "zstd" =>
+ val compressLargerThan =
conf.getBytes("compression.compress-larger-than")
+ val level = conf.getInt("compression.zstd-level")
+ ZstdCompression.checkLevel(level)
+ Compression.Zstd(compressLargerThan, level)
case other =>
throw new IllegalArgumentException(
s"Unknown compression algorithm [$other], possible values are " +
- """"off" or "gzip"""")
+ """"off", "gzip", "lz4" or "zstd"""")
}
}
// "unlimited" or a negative number means no limit; getBytes refuses both,
so read them first
@@ -548,25 +587,35 @@ import pekko.util.OptionVal
case Compression.Off => bytes
case Compression.GZip(largerThan) if bytes.length <= largerThan => bytes
case Compression.GZip(_) =>
- val bos = new ByteArrayOutputStream(BufferSize)
- val zip = new GZIPOutputStream(bos)
+ val bos = new
ByteArrayOutputStream(initialCompressBufferSize(bytes.length))
+ val zip = new GZIPOutputStream(bos, BufferSize)
try zip.write(bytes)
finally zip.close()
bos.toByteArray
case Compression.LZ4(largerThan) if bytes.length <= largerThan => bytes
case Compression.LZ4(_) => {
+ // compress directly after the header, so the output is only copied
once, to trim it
val meta = LZ4Meta(bytes)
- val compressed = lz4Compressor.compress(bytes)
- meta.prependTo(compressed)
+ val maxCompressedLength =
lz4Compressor.maxCompressedLength(bytes.length)
+ val out = new Array[Byte](meta.offset + maxCompressedLength)
+ meta.putInto(ByteBuffer.wrap(out))
+ val compressedLength =
+ lz4Compressor.compress(bytes, 0, bytes.length, out, meta.offset,
maxCompressedLength)
+ java.util.Arrays.copyOf(out, meta.offset + compressedLength)
}
+ case Compression.Zstd(largerThan, _) if bytes.length <= largerThan =>
bytes
+ case Compression.Zstd(_, level) =>
ZstdCompression.compress(bytes, level)
}
}
def decompress(bytes: Array[Byte]): Array[Byte] = {
if (isGZipped(bytes)) {
- val in = new GZIPInputStream(new
UnsynchronizedByteArrayInputStream(bytes))
- try gunzip(in)
+ val in = new GZIPInputStream(new
UnsynchronizedByteArrayInputStream(bytes), BufferSize)
+ try gunzip(in, initialDecompressBufferSize(bytes.length,
maxDecompressedSize))
finally in.close()
+ } else if (isZstd(bytes)) {
+ ZstdCompression.decompress(bytes, maxDecompressedSize,
+ "pekko.serialization.jackson.compression.max-decompressed-size")
} else {
LZ4Meta.get(bytes) match {
case OptionVal.Some(meta) =>
@@ -589,8 +638,8 @@ import pekko.util.OptionVal
// gunzip with a bound on the decompressed size, so a small gzip payload
cannot
// inflate without limit (a "zip bomb"). A negative maximum applies no bound.
- private def gunzip(in: GZIPInputStream): Array[Byte] = {
- val out = new ByteArrayOutputStream()
+ private def gunzip(in: GZIPInputStream, initialBufferSize: Int): Array[Byte]
= {
+ val out = new ByteArrayOutputStream(initialBufferSize)
val buffer = new Array[Byte](BufferSize)
var total = 0L
var n = in.read(buffer)
diff --git
a/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/ZstdCompression.scala
b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/ZstdCompression.scala
new file mode 100644
index 0000000000..2de6ea1ee9
--- /dev/null
+++
b/serialization-jackson/src/main/scala/org/apache/pekko/serialization/jackson/ZstdCompression.scala
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.serialization.jackson
+
+import java.io.ByteArrayOutputStream
+
+import com.github.luben.zstd.{ Zstd, ZstdInputStreamNoFinalizer }
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.io.UnsynchronizedByteArrayInputStream
+
+/**
+ * INTERNAL API
+ *
+ * zstd compression of serialized payloads. zstd-jni is an optional
dependency, so it is only
+ * referenced from this object and only used when zstd is configured or a zstd
payload is
+ * deserialized. A missing zstd-jni (or a platform without its native library)
is reported as an
+ * `IllegalArgumentException` rather than a `LinkageError`.
+ */
+@InternalApi private[jackson] object ZstdCompression {
+
+ private final val MaxArraySize = Int.MaxValue - 8
+
+ private def withZstd[T](action: String)(block: => T): T =
+ try block
+ catch {
+ case e: LinkageError =>
+ throw new IllegalArgumentException(
+ s"zstd is required to $action but is not available. Add the optional
dependency " +
+ "com.github.luben:zstd-jni to the classpath.",
+ e)
+ }
+
+ /** Fails if zstd-jni is not available or `level` is not a valid zstd
compression level. */
+ def checkLevel(level: Int): Unit = withZstd("compress payloads") {
+ if (level < Zstd.minCompressionLevel() || level >
Zstd.maxCompressionLevel())
+ throw new IllegalArgumentException(
+ s"Invalid zstd compression level [$level], it must be between
[${Zstd.minCompressionLevel()}] " +
+ s"and [${Zstd.maxCompressionLevel()}]")
+ }
+
+ def compress(bytes: Array[Byte], level: Int): Array[Byte] =
+ withZstd("compress payloads") {
+ Zstd.compress(bytes, level)
+ }
+
+ /**
+ * Decompress a zstd payload, failing with an `IllegalArgumentException`
once it would exceed
+ * `maxDecompressedSize` bytes, so a small payload cannot expand without
limit. A negative
+ * `maxDecompressedSize` applies no limit. `settingPath` is named in the
error message.
+ */
+ def decompress(bytes: Array[Byte], maxDecompressedSize: Long, settingPath:
String): Array[Byte] =
+ withZstd("deserialize a zstd compressed payload") {
+ // Frames written by `compress` declare their decompressed size. Frames
from other writers,
+ // e.g. a streaming compressor, may not (0 or negative here), and are
decompressed in chunks.
+ val declaredSize = Zstd.getFrameContentSize(bytes)
+ if (declaredSize > 0) {
+ if (maxDecompressedSize >= 0 && declaredSize > maxDecompressedSize)
+ throw new IllegalArgumentException(
+ s"Compressed message declares decompressed size [$declaredSize]
bytes, which exceeds the maximum " +
+ s"of [$maxDecompressedSize] bytes ($settingPath)")
+ if (declaredSize > MaxArraySize)
+ throw new IllegalArgumentException(
+ s"Compressed message declares decompressed size [$declaredSize]
bytes, which is too large")
+ // The declared size is the capacity of the output, so a frame that
produces more than it
+ // declares fails instead of growing beyond the checked size.
+ Zstd.decompress(bytes, declaredSize.toInt)
+ } else decompressStream(bytes, maxDecompressedSize, settingPath)
+ }
+
+ private def decompressStream(bytes: Array[Byte], maxDecompressedSize: Long,
settingPath: String): Array[Byte] = {
+ val in = new ZstdInputStreamNoFinalizer(new
UnsynchronizedByteArrayInputStream(bytes))
+ try {
+ val out =
+ new
ByteArrayOutputStream(JacksonSerializer.initialDecompressBufferSize(bytes.length,
maxDecompressedSize))
+ val buffer = new
Array[Byte](ZstdInputStreamNoFinalizer.recommendedDOutSize().toInt)
+ var total = 0L
+ var n = in.read(buffer)
+ while (n != -1) {
+ total += n
+ if (maxDecompressedSize >= 0 && total > maxDecompressedSize)
+ throw new IllegalArgumentException(
+ s"Decompressed message exceeds the maximum of
[$maxDecompressedSize] bytes ($settingPath)")
+ out.write(buffer, 0, n)
+ n = in.read(buffer)
+ }
+ out.toByteArray
+ } finally in.close()
+ }
+}
diff --git
a/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerBufferSizeSpec.scala
b/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerBufferSizeSpec.scala
new file mode 100644
index 0000000000..09d8d7bf5b
--- /dev/null
+++
b/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerBufferSizeSpec.scala
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.serialization.jackson
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class JacksonSerializerBufferSizeSpec extends AnyWordSpec with Matchers {
+ import JacksonSerializer._
+
+ "initialDecompressBufferSize" should {
+ "be 4x the compressed size" in {
+ initialDecompressBufferSize(10000, -1) should ===(40000)
+ }
+ "be at least 4 KiB" in {
+ initialDecompressBufferSize(100, -1) should ===(4096)
+ initialDecompressBufferSize(10000, 1024) should ===(4096)
+ }
+ "not exceed what the maximum decompressed size allows" in {
+ initialDecompressBufferSize(100000, 50000) should ===(50001)
+ }
+ "be at most 16 MiB" in {
+ initialDecompressBufferSize(10 * 1024 * 1024, -1) should ===(16 * 1024 *
1024)
+ initialDecompressBufferSize(Int.MaxValue, -1) should ===(16 * 1024 *
1024)
+ }
+ }
+
+ "initialCompressBufferSize" should {
+ "be a quarter of the uncompressed size, but at least 4 KiB" in {
+ initialCompressBufferSize(100000) should ===(25000)
+ initialCompressBufferSize(1000) should ===(4096)
+ }
+ }
+}
diff --git
a/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerSpec.scala
b/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerSpec.scala
index 6701651dc3..4f3b3a460e 100644
---
a/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerSpec.scala
+++
b/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonSerializerSpec.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.serialization.jackson
import java.lang
+import java.io.ByteArrayOutputStream
import java.io.NotSerializableException
import java.nio.charset.StandardCharsets
import java.time.Duration
@@ -30,6 +31,8 @@ import scala.annotation.nowarn
import scala.concurrent.duration._
import scala.concurrent.duration.FiniteDuration
+import com.github.luben.zstd.{ Zstd, ZstdOutputStream }
+import net.jpountz.lz4.LZ4Factory
import com.fasterxml.jackson.annotation.JsonIgnore
import com.fasterxml.jackson.annotation.JsonSubTypes
import com.fasterxml.jackson.annotation.JsonTypeInfo
@@ -666,6 +669,24 @@ class JacksonJsonSerializerSpec extends
JacksonSerializerSpec("jackson-json") {
check(SimpleCommand("0" * (compressLargerThan + 1).toInt), true)
}
+ "write lz4 payloads in the LZ4Meta header + lz4 block format" in
withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = lz4
+ compress-larger-than = 0 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val blob = serializeToBinary(msg, sys)
+ val json = withSystem("""
+ pekko.serialization.jackson.jackson-json.compression.algorithm = off
+ """) { plainSys =>
+ serializeToBinary(msg, plainSys)
+ }
+ val expected =
+
JacksonSerializer.LZ4Meta(json).prependTo(LZ4Factory.fastestInstance().fastCompressor().compress(json))
+ blob.toSeq should ===(expected.toSeq)
+ }
+
"not compress small payload with lz4" in withSystem("""
pekko.serialization.jackson.jackson-json.compression {
algorithm = lz4
@@ -753,6 +774,146 @@ class JacksonJsonSerializerSpec extends
JacksonSerializerSpec("jackson-json") {
JacksonSerializer.isGZipped(serializeToBinary(msg, sys)) should ===(true)
checkSerialization(msg, sys)
}
+
+ "compress large payload with zstd" in withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 32 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (33 * 1024))
+ val bytes = serializeToBinary(msg, sys)
+ JacksonSerializer.isZstd(bytes) should ===(true)
+ bytes.length should be < 32 * 1024
+ checkSerialization(msg, sys)
+ }
+
+ "not compress small payload with zstd" in withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 32 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("Bob")
+ JacksonSerializer.isZstd(serializeToBinary(msg, sys)) should ===(false)
+ checkSerialization(msg, sys)
+ }
+
+ "use the configured zstd-level" in {
+ def compressedSize(level: Int): Int =
+ withSystem(s"""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 0 KiB
+ zstd-level = $level
+ }
+ """) { sys =>
+ serializeToBinary(SimpleCommand((1 to 5000).map(i =>
s"item-$i").mkString(",")), sys).length
+ }
+ compressedSize(19) should be < compressedSize(-5)
+ }
+
+ "reject an invalid zstd-level" in {
+ val ex = intercept[IllegalArgumentException] {
+ withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = zstd
+ zstd-level = 100
+ }
+ """) { sys =>
+ serializeToBinary(SimpleCommand("Bob"), sys)
+ }
+ }
+ ex.getMessage should include("Invalid zstd compression level [100]")
+ }
+
+ "deserialize a zstd payload when zstd is not the configured algorithm" in {
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg)
+ val blob = withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 0 KiB
+ }
+ """) { sys =>
+ serializeToBinary(msg, sys)
+ }
+ JacksonSerializer.isZstd(blob) should ===(true)
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg)) should ===(msg)
+ }
+
+ "reject a zstd payload that declares a size beyond max-decompressed-size"
in withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 0 KiB
+ max-decompressed-size = 1 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = serializeToBinary(msg, sys)
+ JacksonSerializer.isZstd(blob) should ===(true)
+ Zstd.getFrameContentSize(blob) should be > 1024L
+ val ex = intercept[IllegalArgumentException] {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ ex.getMessage should include("max-decompressed-size")
+ }
+
+ "reject a zstd payload without declared size that decompresses beyond
max-decompressed-size" in withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ max-decompressed-size = 1 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = zstdStream(serializeToBinary(msg, sys))
+ JacksonSerializer.isZstd(blob) should ===(true)
+ Zstd.getFrameContentSize(blob) should be <= 0L
+ val ex = intercept[IllegalArgumentException] {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ ex.getMessage should include("max-decompressed-size")
+ }
+
+ "deserialize a zstd payload without declared size within
max-decompressed-size" in withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ max-decompressed-size = 1 MiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = zstdStream(serializeToBinary(msg, sys))
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys) should ===(msg)
+ }
+
+ "reject a zstd payload that produces more than its declared size" in
withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ max-decompressed-size = 1 MiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val json = serializeToBinary(msg, sys)
+ // the first frame declares a small size, the frame appended to it
decompresses to much more
+ val blob = Zstd.compress(json.take(10)) ++ Zstd.compress(json.drop(10))
+ Zstd.getFrameContentSize(blob) should ===(10L)
+ an[Exception] should be thrownBy {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ }
+
+ "apply no zstd decompression limit when max-decompressed-size is -1" in
withSystem("""
+ pekko.serialization.jackson.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 0 KiB
+ max-decompressed-size = -1
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ JacksonSerializer.isZstd(serializeToBinary(msg, sys)) should ===(true)
+ checkSerialization(msg, sys)
+ }
}
"JacksonJsonSerializer without type in manifest" should {
@@ -897,7 +1058,7 @@ abstract class JacksonSerializerSpec(serializerName:
String)
if (blob.length > 0) {
serializer match {
case _: JacksonJsonSerializer =>
- if (!JacksonSerializer.isGZipped(blob) &&
!JacksonSerializer.isLZ4(blob))
+ if (!JacksonSerializer.isGZipped(blob) &&
!JacksonSerializer.isLZ4(blob) && !JacksonSerializer.isZstd(blob))
new String(blob.take(1), StandardCharsets.UTF_8) should ===("{")
case _: JacksonCborSerializer =>
new String(blob.take(1), StandardCharsets.UTF_8) should !==("{")
@@ -910,6 +1071,15 @@ abstract class JacksonSerializerSpec(serializerName:
String)
deserialized should ===(obj)
}
+ // zstd compress with the streaming API, so the frame does not declare its
decompressed size
+ def zstdStream(bytes: Array[Byte]): Array[Byte] = {
+ val bos = new ByteArrayOutputStream()
+ val out = new ZstdOutputStream(bos)
+ try out.write(bytes)
+ finally out.close()
+ bos.toByteArray
+ }
+
def serializeToBinary(obj: AnyRef, sys: ActorSystem = system): Array[Byte] =
serialization(sys).serialize(obj).get
diff --git
a/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonWithoutZstdSpec.scala
b/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonWithoutZstdSpec.scala
new file mode 100644
index 0000000000..84301ff8c2
--- /dev/null
+++
b/serialization-jackson/src/test/scala/org/apache/pekko/serialization/jackson/JacksonWithoutZstdSpec.scala
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.serialization.jackson
+
+import java.io.File
+import java.net.URLClassLoader
+import java.nio.charset.StandardCharsets
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+import scala.util.control.NonFatal
+
+import com.github.luben.zstd.Zstd
+
+import org.apache.pekko
+import pekko.actor.{ ActorSystem, ExtendedActorSystem }
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import com.typesafe.config.ConfigFactory
+
+/**
+ * zstd-jni is an optional dependency, so the Jackson serializers must work
without it and fail
+ * with a clear error when zstd is configured or a zstd payload is received.
+ */
+class JacksonWithoutZstdSpec extends AnyWordSpec with Matchers {
+
+ "The Jackson serializer" should {
+ "work without zstd-jni on the classpath and report zstd use clearly" in {
+ // tests run in a forked JVM, so java.class.path is the test classpath
+ val urls = System
+ .getProperty("java.class.path")
+ .split(File.pathSeparator)
+ .filterNot(new File(_).getName.startsWith("zstd-jni"))
+ .map(new File(_).toURI.toURL)
+ // not closed: dispatcher shutdown tasks of the terminated ActorSystem
may still load classes from it
+ val loader = new URLClassLoader(urls, ClassLoader.getPlatformClassLoader)
+ a[ClassNotFoundException] should be thrownBy
loader.loadClass("com.github.luben.zstd.Zstd")
+
+ val zstdPayload =
Zstd.compress("{\"name\":\"Bob\"}".getBytes(StandardCharsets.UTF_8))
+ val check = loader.loadClass(classOf[JacksonWithoutZstdCheck].getName)
+ val thread = Thread.currentThread()
+ val contextClassLoader = thread.getContextClassLoader
+ thread.setContextClassLoader(loader)
+ val result =
+ try
+ check
+ .getMethod("run", classOf[Array[Byte]])
+ .invoke(check.getDeclaredConstructor().newInstance(), zstdPayload)
+ finally thread.setContextClassLoader(contextClassLoader)
+ result should ===(
+ "gzip:Bob zstd-algorithm:IllegalArgumentException(zstd-jni)
zstd-payload:IllegalArgumentException(zstd-jni)")
+ }
+ }
+}
+
+/** Loaded and run in a class loader without zstd-jni by
[[JacksonWithoutZstdSpec]]. */
+class JacksonWithoutZstdCheck {
+ def run(zstdPayload: Array[Byte]): String = {
+ // use this class loader (without zstd-jni) rather than one found from the
context or call stack
+ val classLoader = getClass.getClassLoader
+ // a fatal error in this nested ActorSystem must fail the test, not exit
the forked test JVM
+ val config = ConfigFactory
+ .parseString("""
+ pekko.jvm-exit-on-fatal-error = off
+
pekko.serialization.jackson.jackson-json.compression.compress-larger-than = 0
KiB
+ pekko.serialization.jackson.allowed-class-prefix =
["org.apache.pekko.serialization.jackson."]
+ pekko.serialization.jackson.jackson-zstd.compression {
+ algorithm = zstd
+ }
+ """)
+ .withFallback(ConfigFactory.load(classLoader))
+ val system = ActorSystem("JacksonWithoutZstd", Some(config),
Some(classLoader), None)
+ try {
+ val extendedSystem = system.asInstanceOf[ExtendedActorSystem]
+ def describe(e: Throwable): String =
+ s"${e.getClass.getSimpleName}(${if (e.getMessage.contains("zstd-jni"))
"zstd-jni" else e.getMessage})"
+
+ val json = new JacksonJsonSerializer(extendedSystem, "jackson-json")
+ val msg = ScalaTestMessages.SimpleCommand("Bob")
+ val manifest = json.manifest(msg)
+ val gzip = json.fromBinary(json.toBinary(msg), manifest) match {
+ case ScalaTestMessages.SimpleCommand(name) => name
+ case other => other.toString
+ }
+
+ val zstdAlgorithm =
+ try {
+ new JacksonJsonSerializer(extendedSystem, "jackson-zstd")
+ "created"
+ } catch { case NonFatal(e) => describe(e) }
+
+ val zstdPayloadResult =
+ try json.fromBinary(zstdPayload, manifest).toString
+ catch { case NonFatal(e) => describe(e) }
+
+ s"gzip:$gzip zstd-algorithm:$zstdAlgorithm
zstd-payload:$zstdPayloadResult"
+ } finally {
+ system.terminate()
+ Await.ready(system.whenTerminated, 10.seconds)
+ }
+ }
+}
diff --git a/serialization-jackson3/src/main/resources/reference.conf
b/serialization-jackson3/src/main/resources/reference.conf
index 35c550633f..8b211fb2e5 100644
--- a/serialization-jackson3/src/main/resources/reference.conf
+++ b/serialization-jackson3/src/main/resources/reference.conf
@@ -181,13 +181,19 @@ pekko.serialization.jackson3 {
# Compression algorithm.
# - off : no compression
# - gzip : using common java gzip
+ # - lz4 : using lz4-java
+ # - zstd : using zstd-jni, an optional dependency that has to be added to
the classpath
algorithm = off
# If compression is enabled with the `algorithm` setting the payload is
compressed
# when it's larger than this value.
compress-larger-than = 0 KiB
- # Maximum size of a payload after decompression. A compressed (gzip or lz4)
+ # zstd compression level used when `algorithm` is zstd, from -131072
(fastest) to
+ # 22 (strongest compression). 3 is zstd's default level.
+ zstd-level = 3
+
+ # Maximum size of a payload after decompression. A compressed (gzip, lz4
or zstd)
# payload that decompresses to more than this is rejected rather than
# allocated, guarding against a small message that inflates without bound.
# This applies on deserialization regardless of the `algorithm` setting
above.
@@ -243,6 +249,7 @@ pekko.serialization.jackson3.jackson-json.compression {
# - off : no compression
# - gzip : using common java gzip
# - lz4 : using lz4-java
+ # - zstd : using zstd-jni, an optional dependency that has to be added to
the classpath
algorithm = gzip
# If compression is enabled with the `algorithm` setting the payload is
compressed
diff --git
a/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala
b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala
index c901176dc9..cf0e1f0f27 100644
---
a/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala
+++
b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/JacksonSerializer.scala
@@ -140,6 +140,39 @@ import pekko.util.OptionVal
LZ4Meta.get(bytes).isDefined
}
+ private final val MinBufferSize = 4 * 1024
+ private final val MaxInitialDecompressBufferSize = 16 * 1024 * 1024
+
+ /**
+ * Initial size of the buffer that a compressed payload is decompressed into
when its
+ * decompressed size is not known up front. JSON and CBOR typically compress
3-10x, so 4x the
+ * compressed size avoids most resizing without allocating much more than
needed. It is never
+ * larger than the configured maximum decompressed size allows, and capped
so that a large
+ * payload grows the buffer rather than allocating a huge one up front.
+ */
+ def initialDecompressBufferSize(compressedSize: Int, maxDecompressedSize:
Long): Int = {
+ val estimate = compressedSize.toLong * 4
+ val bounded = if (maxDecompressedSize >= 0) math.min(estimate,
maxDecompressedSize + 1) else estimate
+ math.max(MinBufferSize, math.min(bounded,
MaxInitialDecompressBufferSize.toLong)).toInt
+ }
+
+ /**
+ * Initial size of the buffer that a payload is compressed into: a quarter
of its size (JSON and
+ * CBOR typically compress at least that well), but at least 4 KiB.
+ */
+ def initialCompressBufferSize(uncompressedSize: Int): Int =
+ math.max(MinBufferSize, uncompressedSize / 4)
+
+ // a zstd frame starts with the magic number 0xFD2FB528, stored little-endian
+ def isZstd(bytes: Array[Byte]): Boolean = {
+ (bytes ne null) &&
+ (bytes.length >= 4) &&
+ (bytes(0) == 0x28.toByte) &&
+ (bytes(1) == 0xB5.toByte) &&
+ (bytes(2) == 0x2F.toByte) &&
+ (bytes(3) == 0xFD.toByte)
+ }
+
}
/**
@@ -169,6 +202,7 @@ import pekko.util.OptionVal
object Off extends Algorithm
final case class GZip(largerThan: Long) extends Algorithm
final case class LZ4(largerThan: Long) extends Algorithm
+ final case class Zstd(largerThan: Long, level: Int) extends Algorithm
}
/**
@@ -203,10 +237,15 @@ import pekko.util.OptionVal
case "lz4" =>
val compressLargerThan =
conf.getBytes("compression.compress-larger-than")
Compression.LZ4(compressLargerThan)
+ case "zstd" =>
+ val compressLargerThan =
conf.getBytes("compression.compress-larger-than")
+ val level = conf.getInt("compression.zstd-level")
+ ZstdCompression.checkLevel(level)
+ Compression.Zstd(compressLargerThan, level)
case other =>
throw new IllegalArgumentException(
s"Unknown compression algorithm [$other], possible values are " +
- """"off" or "gzip"""")
+ """"off", "gzip", "lz4" or "zstd"""")
}
}
// "unlimited" or a negative number means no limit; getBytes refuses both,
so read them first
@@ -549,25 +588,35 @@ import pekko.util.OptionVal
case Compression.Off => bytes
case Compression.GZip(largerThan) if bytes.length <= largerThan => bytes
case Compression.GZip(_) =>
- val bos = new ByteArrayOutputStream(BufferSize)
- val zip = new GZIPOutputStream(bos)
+ val bos = new
ByteArrayOutputStream(initialCompressBufferSize(bytes.length))
+ val zip = new GZIPOutputStream(bos, BufferSize)
try zip.write(bytes)
finally zip.close()
bos.toByteArray
case Compression.LZ4(largerThan) if bytes.length <= largerThan => bytes
case Compression.LZ4(_) => {
+ // compress directly after the header, so the output is only copied
once, to trim it
val meta = LZ4Meta(bytes)
- val compressed = lz4Compressor.compress(bytes)
- meta.prependTo(compressed)
+ val maxCompressedLength =
lz4Compressor.maxCompressedLength(bytes.length)
+ val out = new Array[Byte](meta.offset + maxCompressedLength)
+ meta.putInto(ByteBuffer.wrap(out))
+ val compressedLength =
+ lz4Compressor.compress(bytes, 0, bytes.length, out, meta.offset,
maxCompressedLength)
+ java.util.Arrays.copyOf(out, meta.offset + compressedLength)
}
+ case Compression.Zstd(largerThan, _) if bytes.length <= largerThan =>
bytes
+ case Compression.Zstd(_, level) =>
ZstdCompression.compress(bytes, level)
}
}
def decompress(bytes: Array[Byte]): Array[Byte] = {
if (isGZipped(bytes)) {
- val in = new GZIPInputStream(new
UnsynchronizedByteArrayInputStream(bytes))
- try gunzip(in)
+ val in = new GZIPInputStream(new
UnsynchronizedByteArrayInputStream(bytes), BufferSize)
+ try gunzip(in, initialDecompressBufferSize(bytes.length,
maxDecompressedSize))
finally in.close()
+ } else if (isZstd(bytes)) {
+ ZstdCompression.decompress(bytes, maxDecompressedSize,
+ "pekko.serialization.jackson3.compression.max-decompressed-size")
} else {
LZ4Meta.get(bytes) match {
case OptionVal.Some(meta) =>
@@ -590,8 +639,8 @@ import pekko.util.OptionVal
// gunzip with a bound on the decompressed size, so a small gzip payload
cannot
// inflate without limit (a "zip bomb"). A negative maximum applies no bound.
- private def gunzip(in: GZIPInputStream): Array[Byte] = {
- val out = new ByteArrayOutputStream()
+ private def gunzip(in: GZIPInputStream, initialBufferSize: Int): Array[Byte]
= {
+ val out = new ByteArrayOutputStream(initialBufferSize)
val buffer = new Array[Byte](BufferSize)
var total = 0L
var n = in.read(buffer)
diff --git
a/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/ZstdCompression.scala
b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/ZstdCompression.scala
new file mode 100644
index 0000000000..a7dad2ea39
--- /dev/null
+++
b/serialization-jackson3/src/main/scala/org/apache/pekko/serialization/jackson3/ZstdCompression.scala
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.serialization.jackson3
+
+import java.io.ByteArrayOutputStream
+
+import com.github.luben.zstd.{ Zstd, ZstdInputStreamNoFinalizer }
+
+import org.apache.pekko
+import pekko.annotation.InternalApi
+import pekko.io.UnsynchronizedByteArrayInputStream
+
+/**
+ * INTERNAL API
+ *
+ * zstd compression of serialized payloads. zstd-jni is an optional
dependency, so it is only
+ * referenced from this object and only used when zstd is configured or a zstd
payload is
+ * deserialized. A missing zstd-jni (or a platform without its native library)
is reported as an
+ * `IllegalArgumentException` rather than a `LinkageError`.
+ */
+@InternalApi private[jackson3] object ZstdCompression {
+
+ private final val MaxArraySize = Int.MaxValue - 8
+
+ private def withZstd[T](action: String)(block: => T): T =
+ try block
+ catch {
+ case e: LinkageError =>
+ throw new IllegalArgumentException(
+ s"zstd is required to $action but is not available. Add the optional
dependency " +
+ "com.github.luben:zstd-jni to the classpath.",
+ e)
+ }
+
+ /** Fails if zstd-jni is not available or `level` is not a valid zstd
compression level. */
+ def checkLevel(level: Int): Unit = withZstd("compress payloads") {
+ if (level < Zstd.minCompressionLevel() || level >
Zstd.maxCompressionLevel())
+ throw new IllegalArgumentException(
+ s"Invalid zstd compression level [$level], it must be between
[${Zstd.minCompressionLevel()}] " +
+ s"and [${Zstd.maxCompressionLevel()}]")
+ }
+
+ def compress(bytes: Array[Byte], level: Int): Array[Byte] =
+ withZstd("compress payloads") {
+ Zstd.compress(bytes, level)
+ }
+
+ /**
+ * Decompress a zstd payload, failing with an `IllegalArgumentException`
once it would exceed
+ * `maxDecompressedSize` bytes, so a small payload cannot expand without
limit. A negative
+ * `maxDecompressedSize` applies no limit. `settingPath` is named in the
error message.
+ */
+ def decompress(bytes: Array[Byte], maxDecompressedSize: Long, settingPath:
String): Array[Byte] =
+ withZstd("deserialize a zstd compressed payload") {
+ // Frames written by `compress` declare their decompressed size. Frames
from other writers,
+ // e.g. a streaming compressor, may not (0 or negative here), and are
decompressed in chunks.
+ val declaredSize = Zstd.getFrameContentSize(bytes)
+ if (declaredSize > 0) {
+ if (maxDecompressedSize >= 0 && declaredSize > maxDecompressedSize)
+ throw new IllegalArgumentException(
+ s"Compressed message declares decompressed size [$declaredSize]
bytes, which exceeds the maximum " +
+ s"of [$maxDecompressedSize] bytes ($settingPath)")
+ if (declaredSize > MaxArraySize)
+ throw new IllegalArgumentException(
+ s"Compressed message declares decompressed size [$declaredSize]
bytes, which is too large")
+ // The declared size is the capacity of the output, so a frame that
produces more than it
+ // declares fails instead of growing beyond the checked size.
+ Zstd.decompress(bytes, declaredSize.toInt)
+ } else decompressStream(bytes, maxDecompressedSize, settingPath)
+ }
+
+ private def decompressStream(bytes: Array[Byte], maxDecompressedSize: Long,
settingPath: String): Array[Byte] = {
+ val in = new ZstdInputStreamNoFinalizer(new
UnsynchronizedByteArrayInputStream(bytes))
+ try {
+ val out =
+ new
ByteArrayOutputStream(JacksonSerializer.initialDecompressBufferSize(bytes.length,
maxDecompressedSize))
+ val buffer = new
Array[Byte](ZstdInputStreamNoFinalizer.recommendedDOutSize().toInt)
+ var total = 0L
+ var n = in.read(buffer)
+ while (n != -1) {
+ total += n
+ if (maxDecompressedSize >= 0 && total > maxDecompressedSize)
+ throw new IllegalArgumentException(
+ s"Decompressed message exceeds the maximum of
[$maxDecompressedSize] bytes ($settingPath)")
+ out.write(buffer, 0, n)
+ n = in.read(buffer)
+ }
+ out.toByteArray
+ } finally in.close()
+ }
+}
diff --git
a/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerBufferSizeSpec.scala
b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerBufferSizeSpec.scala
new file mode 100644
index 0000000000..04afbac852
--- /dev/null
+++
b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerBufferSizeSpec.scala
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.serialization.jackson3
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+class JacksonSerializerBufferSizeSpec extends AnyWordSpec with Matchers {
+ import JacksonSerializer._
+
+ "initialDecompressBufferSize" should {
+ "be 4x the compressed size" in {
+ initialDecompressBufferSize(10000, -1) should ===(40000)
+ }
+ "be at least 4 KiB" in {
+ initialDecompressBufferSize(100, -1) should ===(4096)
+ initialDecompressBufferSize(10000, 1024) should ===(4096)
+ }
+ "not exceed what the maximum decompressed size allows" in {
+ initialDecompressBufferSize(100000, 50000) should ===(50001)
+ }
+ "be at most 16 MiB" in {
+ initialDecompressBufferSize(10 * 1024 * 1024, -1) should ===(16 * 1024 *
1024)
+ initialDecompressBufferSize(Int.MaxValue, -1) should ===(16 * 1024 *
1024)
+ }
+ }
+
+ "initialCompressBufferSize" should {
+ "be a quarter of the uncompressed size, but at least 4 KiB" in {
+ initialCompressBufferSize(100000) should ===(25000)
+ initialCompressBufferSize(1000) should ===(4096)
+ }
+ }
+}
diff --git
a/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
index dff3deca43..9d158d4175 100644
---
a/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
+++
b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonSerializerSpec.scala
@@ -14,6 +14,7 @@
package org.apache.pekko.serialization.jackson3
import java.lang
+import java.io.ByteArrayOutputStream
import java.io.NotSerializableException
import java.nio.charset.StandardCharsets
import java.time.{ Duration, Instant, LocalDateTime }
@@ -25,6 +26,8 @@ import scala.annotation.nowarn
import scala.concurrent.duration._
import scala.concurrent.duration.FiniteDuration
+import com.github.luben.zstd.{ Zstd, ZstdOutputStream }
+import net.jpountz.lz4.LZ4Factory
import com.fasterxml.jackson.annotation.{ JsonIgnore, JsonSubTypes,
JsonTypeInfo }
import tools.jackson.core.{ JsonGenerator, StreamReadFeature,
StreamWriteFeature }
import tools.jackson.core.`type`.TypeReference
@@ -611,6 +614,24 @@ class JacksonJsonSerializerSpec extends
JacksonSerializerSpec("jackson-json") {
check(SimpleCommand("0" * (compressLargerThan + 1).toInt), true)
}
+ "write lz4 payloads in the LZ4Meta header + lz4 block format" in
withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = lz4
+ compress-larger-than = 0 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val blob = serializeToBinary(msg, sys)
+ val json = withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression.algorithm = off
+ """) { plainSys =>
+ serializeToBinary(msg, plainSys)
+ }
+ val expected =
+
JacksonSerializer.LZ4Meta(json).prependTo(LZ4Factory.fastestInstance().fastCompressor().compress(json))
+ blob.toSeq should ===(expected.toSeq)
+ }
+
"not compress small payload with lz4" in withSystem("""
pekko.serialization.jackson3.jackson-json.compression {
algorithm = lz4
@@ -698,6 +719,146 @@ class JacksonJsonSerializerSpec extends
JacksonSerializerSpec("jackson-json") {
JacksonSerializer.isGZipped(serializeToBinary(msg, sys)) should ===(true)
checkSerialization(msg, sys)
}
+
+ "compress large payload with zstd" in withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 32 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (33 * 1024))
+ val bytes = serializeToBinary(msg, sys)
+ JacksonSerializer.isZstd(bytes) should ===(true)
+ bytes.length should be < 32 * 1024
+ checkSerialization(msg, sys)
+ }
+
+ "not compress small payload with zstd" in withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 32 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("Bob")
+ JacksonSerializer.isZstd(serializeToBinary(msg, sys)) should ===(false)
+ checkSerialization(msg, sys)
+ }
+
+ "use the configured zstd-level" in {
+ def compressedSize(level: Int): Int =
+ withSystem(s"""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 0 KiB
+ zstd-level = $level
+ }
+ """) { sys =>
+ serializeToBinary(SimpleCommand((1 to 5000).map(i =>
s"item-$i").mkString(",")), sys).length
+ }
+ compressedSize(19) should be < compressedSize(-5)
+ }
+
+ "reject an invalid zstd-level" in {
+ val ex = intercept[IllegalArgumentException] {
+ withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = zstd
+ zstd-level = 100
+ }
+ """) { sys =>
+ serializeToBinary(SimpleCommand("Bob"), sys)
+ }
+ }
+ ex.getMessage should include("Invalid zstd compression level [100]")
+ }
+
+ "deserialize a zstd payload when zstd is not the configured algorithm" in {
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg)
+ val blob = withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 0 KiB
+ }
+ """) { sys =>
+ serializeToBinary(msg, sys)
+ }
+ JacksonSerializer.isZstd(blob) should ===(true)
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg)) should ===(msg)
+ }
+
+ "reject a zstd payload that declares a size beyond max-decompressed-size"
in withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 0 KiB
+ max-decompressed-size = 1 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = serializeToBinary(msg, sys)
+ JacksonSerializer.isZstd(blob) should ===(true)
+ Zstd.getFrameContentSize(blob) should be > 1024L
+ val ex = intercept[IllegalArgumentException] {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ ex.getMessage should include("max-decompressed-size")
+ }
+
+ "reject a zstd payload without declared size that decompresses beyond
max-decompressed-size" in withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ max-decompressed-size = 1 KiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = zstdStream(serializeToBinary(msg, sys))
+ JacksonSerializer.isZstd(blob) should ===(true)
+ Zstd.getFrameContentSize(blob) should be <= 0L
+ val ex = intercept[IllegalArgumentException] {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ ex.getMessage should include("max-decompressed-size")
+ }
+
+ "deserialize a zstd payload without declared size within
max-decompressed-size" in withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ max-decompressed-size = 1 MiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val blob = zstdStream(serializeToBinary(msg, sys))
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys) should ===(msg)
+ }
+
+ "reject a zstd payload that produces more than its declared size" in
withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ max-decompressed-size = 1 MiB
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ val serializer = serializerFor(msg, sys)
+ val json = serializeToBinary(msg, sys)
+ // the first frame declares a small size, the frame appended to it
decompresses to much more
+ val blob = Zstd.compress(json.take(10)) ++ Zstd.compress(json.drop(10))
+ Zstd.getFrameContentSize(blob) should ===(10L)
+ an[Exception] should be thrownBy {
+ deserializeFromBinary(blob, serializer.identifier,
serializer.manifest(msg), sys)
+ }
+ }
+
+ "apply no zstd decompression limit when max-decompressed-size is -1" in
withSystem("""
+ pekko.serialization.jackson3.jackson-json.compression {
+ algorithm = zstd
+ compress-larger-than = 0 KiB
+ max-decompressed-size = -1
+ }
+ """) { sys =>
+ val msg = SimpleCommand("0" * (8 * 1024))
+ JacksonSerializer.isZstd(serializeToBinary(msg, sys)) should ===(true)
+ checkSerialization(msg, sys)
+ }
}
"JacksonJsonSerializer without type in manifest" should {
@@ -842,7 +1003,7 @@ abstract class JacksonSerializerSpec(serializerName:
String)
if (blob.length > 0) {
serializer match {
case _: JacksonJsonSerializer =>
- if (!JacksonSerializer.isGZipped(blob) &&
!JacksonSerializer.isLZ4(blob))
+ if (!JacksonSerializer.isGZipped(blob) &&
!JacksonSerializer.isLZ4(blob) && !JacksonSerializer.isZstd(blob))
new String(blob.take(1), StandardCharsets.UTF_8) should ===("{")
case _: JacksonCborSerializer =>
new String(blob.take(1), StandardCharsets.UTF_8) should !==("{")
@@ -855,6 +1016,15 @@ abstract class JacksonSerializerSpec(serializerName:
String)
deserialized should ===(obj)
}
+ // zstd compress with the streaming API, so the frame does not declare its
decompressed size
+ def zstdStream(bytes: Array[Byte]): Array[Byte] = {
+ val bos = new ByteArrayOutputStream()
+ val out = new ZstdOutputStream(bos)
+ try out.write(bytes)
+ finally out.close()
+ bos.toByteArray
+ }
+
def serializeToBinary(obj: AnyRef, sys: ActorSystem = system): Array[Byte] =
serialization(sys).serialize(obj).get
diff --git
a/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonWithoutZstdSpec.scala
b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonWithoutZstdSpec.scala
new file mode 100644
index 0000000000..12b5cd6687
--- /dev/null
+++
b/serialization-jackson3/src/test/scala/org/apache/pekko/serialization/jackson3/JacksonWithoutZstdSpec.scala
@@ -0,0 +1,119 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.pekko.serialization.jackson3
+
+import java.io.File
+import java.net.URLClassLoader
+import java.nio.charset.StandardCharsets
+
+import scala.concurrent.Await
+import scala.concurrent.duration._
+import scala.util.control.NonFatal
+
+import com.github.luben.zstd.Zstd
+
+import org.apache.pekko
+import pekko.actor.{ ActorSystem, ExtendedActorSystem }
+
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import com.typesafe.config.ConfigFactory
+
+/**
+ * zstd-jni is an optional dependency, so the Jackson serializers must work
without it and fail
+ * with a clear error when zstd is configured or a zstd payload is received.
+ */
+class JacksonWithoutZstdSpec extends AnyWordSpec with Matchers {
+
+ "The Jackson serializer" should {
+ "work without zstd-jni on the classpath and report zstd use clearly" in {
+ // tests run in a forked JVM, so java.class.path is the test classpath
+ val urls = System
+ .getProperty("java.class.path")
+ .split(File.pathSeparator)
+ .filterNot(new File(_).getName.startsWith("zstd-jni"))
+ .map(new File(_).toURI.toURL)
+ // not closed: dispatcher shutdown tasks of the terminated ActorSystem
may still load classes from it
+ val loader = new URLClassLoader(urls, ClassLoader.getPlatformClassLoader)
+ a[ClassNotFoundException] should be thrownBy
loader.loadClass("com.github.luben.zstd.Zstd")
+
+ val zstdPayload =
Zstd.compress("{\"name\":\"Bob\"}".getBytes(StandardCharsets.UTF_8))
+ val check = loader.loadClass(classOf[JacksonWithoutZstdCheck].getName)
+ val thread = Thread.currentThread()
+ val contextClassLoader = thread.getContextClassLoader
+ thread.setContextClassLoader(loader)
+ val result =
+ try
+ check
+ .getMethod("run", classOf[Array[Byte]])
+ .invoke(check.getDeclaredConstructor().newInstance(), zstdPayload)
+ finally thread.setContextClassLoader(contextClassLoader)
+ result should ===(
+ "gzip:Bob zstd-algorithm:IllegalArgumentException(zstd-jni)
zstd-payload:IllegalArgumentException(zstd-jni)")
+ }
+ }
+}
+
+/** Loaded and run in a class loader without zstd-jni by
[[JacksonWithoutZstdSpec]]. */
+class JacksonWithoutZstdCheck {
+ def run(zstdPayload: Array[Byte]): String = {
+ // use this class loader (without zstd-jni) rather than one found from the
context or call stack
+ val classLoader = getClass.getClassLoader
+ // a fatal error in this nested ActorSystem must fail the test, not exit
the forked test JVM
+ val config = ConfigFactory
+ .parseString("""
+ pekko.jvm-exit-on-fatal-error = off
+
pekko.serialization.jackson3.jackson-json.compression.compress-larger-than = 0
KiB
+ pekko.serialization.jackson3.allowed-class-prefix =
["org.apache.pekko.serialization.jackson3."]
+ pekko.serialization.jackson3.jackson-zstd.compression {
+ algorithm = zstd
+ }
+ """)
+ .withFallback(ConfigFactory.load(classLoader))
+ val system = ActorSystem("JacksonWithoutZstd", Some(config),
Some(classLoader), None)
+ try {
+ val extendedSystem = system.asInstanceOf[ExtendedActorSystem]
+ def describe(e: Throwable): String =
+ s"${e.getClass.getSimpleName}(${if (e.getMessage.contains("zstd-jni"))
"zstd-jni" else e.getMessage})"
+
+ val json = new JacksonJsonSerializer(extendedSystem, "jackson-json")
+ val msg = ScalaTestMessages.SimpleCommand("Bob")
+ val manifest = json.manifest(msg)
+ val gzip = json.fromBinary(json.toBinary(msg), manifest) match {
+ case ScalaTestMessages.SimpleCommand(name) => name
+ case other => other.toString
+ }
+
+ val zstdAlgorithm =
+ try {
+ new JacksonJsonSerializer(extendedSystem, "jackson-zstd")
+ "created"
+ } catch { case NonFatal(e) => describe(e) }
+
+ val zstdPayloadResult =
+ try json.fromBinary(zstdPayload, manifest).toString
+ catch { case NonFatal(e) => describe(e) }
+
+ s"gzip:$gzip zstd-algorithm:$zstdAlgorithm
zstd-payload:$zstdPayloadResult"
+ } finally {
+ system.terminate()
+ Await.ready(system.whenTerminated, 10.seconds)
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
