The GitHub Actions job "Stage release candidate" on pekko.git/x2.0.0-M5-RC0 has 
failed.
Run started by GitHub user pjfanning (triggered by pjfanning).

Head commit for run:
b88cae3d47c8457e8bc49558e52d7be4f120ca9f / PJ Fanning 
<[email protected]>
Add zstd compression and decompression for pekko streams (heap buffer APIs) 
(#3639)

* Add zstd compression and decompression for pekko streams

* Use zstd-jni heap buffer streaming APIs for zstd compression

Motivation:
zstd-jni 1.5.7-23 added ZstdCompressCtx.compressByteBufferStream and
ZstdDecompressCtx.decompressByteBufferStream, which accept array-backed
heap ByteBuffers (luben/zstd-jni#324). The zstd stages no longer need to
copy every element into fixed size direct buffers.

Modification:
- Upgrade zstd-jni to 1.5.7-24.
- Rewrite ZstdCompressor on ZstdCompressCtx and ZstdDecompressor on
  ZstdDecompressCtx using heap buffers.
- Add ZstdBuffers to feed ByteString fragments to zstd and to hand
  output arrays to ByteStrings without copying where possible.
- Fail zstdDecompress on truncated input, emit chunks of at most
  maxBytesPerChunk and pass the dictionary offset, length and level to
  ZstdDictCompress in the right order.
- Revert the ByteBufferCleaner visibility change, it is no longer used.

Result:
zstd compression and decompression avoid direct buffers and most array
copies, and detect truncated streams.

Tests:
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.* 
org.apache.pekko.stream.scaladsl.CompressionSpec": passed
- sbt "stream/mimaReportBinaryIssues": passed

References:
Refs #2404, Refs #2409, Refs luben/zstd-jni#324

* Add ByteString.asByteBuffersUnsafe for zero-copy heap buffers

Motivation:
zstd-jni needs array-backed buffers, but asByteBuffers returns read-only
buffers with no accessible array and code outside ByteString can only get
at a sliced fragment's bytes via toArrayUnsafe, which copies.

Modification:
Add private[pekko] ByteString.asByteBuffersUnsafe, which wraps each
non-empty fragment's backing array (with its start index and length)
without copying, and use it for the zstd input buffers.

Result:
zstd compression and decompression no longer copy sliced or fragmented
input ByteStrings.

Tests:
- sbt "actor-tests/testOnly org.apache.pekko.util.ByteString*": passed
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.* 
org.apache.pekko.stream.scaladsl.CompressionSpec": passed
- sbt "actor/mimaReportBinaryIssues" "stream/mimaReportBinaryIssues": passed

References:
Refs #2409, Refs luben/zstd-jni#324

* Make the zstd-jni dependency of pekko-stream optional

Motivation:
zstd-jni is a 6.5 MB jar of native libraries that most pekko-stream
users do not need. The Compression objects called zstd-jni during their
initialization, so gzip and deflate users loaded the native library and
would fail without zstd-jni on the classpath.

Modification:
- Declare zstd-jni as Optional for pekko-stream (and as an optional
  OSGi import), add it to the stream-tests test classpath.
- Replace the zstd-jni calls for the zstd constants in the scaladsl and
  javadsl Compression objects with libzstd's values, documented in code.
- Document the dependency in the zstd operator docs.

Result:
pekko-stream does not pull in zstd-jni, and the other compression
operators work without it.

Tests:
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.* 
org.apache.pekko.stream.scaladsl.CompressionSpec": passed,
  including a gzip/deflate round trip in a class loader without zstd-jni
  and a check that the constants match zstd-jni
- sbt "stream/mimaReportBinaryIssues": passed
- sbt "stream/makePom" "stream/packageBin": zstd-jni is optional in the
  POM and the OSGi Import-Package

References:
Refs #2409, Refs luben/zstd-jni#324

* Add JMH benchmark for gzip and zstd compression flows

Motivation:
There were no benchmarks for the stream compression operators.

Modification:
Add CompressionBenchmark comparing gzip and zstd compression and
decompression of 16 MiB of JSON-like text in 1 KiB and 64 KiB elements,
and add zstd-jni to bench-jmh (it is optional in pekko-stream).

Result:
The zstd operators can be benchmarked against gzip.

Tests:
- sbt "bench-jmh/Jmh/run -wi 3 -w 2s -i 5 -r 2s -f 1 .*CompressionBenchmark.*": 
ran

References:
Refs #2409

* Only hand over nearly full zstd output buffers without copying

Motivation:
ZstdBuffers.takeOutput handed the output array over to the emitted
ByteString once it was half full, so e.g. 64 KiB decompressed chunks kept
a 128 KiB array alive and every chunk allocated a new 128 KiB array.

Modification:
Hand the array over only when at least 7/8 of it is used, otherwise copy
the output into an exact-size array and reuse the buffer.

Result:
Emitted ByteStrings retain at most 1/8 unused space.

Tests:
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.* 
org.apache.pekko.stream.scaladsl.CompressionSpec": passed

References:
Refs #2409

* Remove the zstd-jni default from ZstdDecompressor

Motivation:
ZstdDecompressor's maxBytesPerChunk defaulted to Zstd.blockSizeMax(), the only
zstd default not taken from the documented Compression constants. The gzip and
deflate decompressor stages have no default, and every caller passes the value.

Modification:
Make maxBytesPerChunk a required parameter of ZstdDecompressor.

Result:
ZstdDecompressor matches the other decompressor stages, and callers pick the
chunk size explicitly (Compression.zstdDecompress already does).

Tests:
sbt "stream/scalafmt" "stream-tests/testOnly 
org.apache.pekko.stream.io.compression.*"
(145 succeeded, 0 failed)

References:
Review comment https://github.com/apache/pekko/pull/3639#issuecomment-6347243585

* Keep the zstd-free class loader open in CompressionWithoutZstdSpec

Closing the URLClassLoader right after the nested ActorSystem terminated
let a late dispatcher shutdown task fail with NoClassDefFoundError, which
jvm-exit-on-fatal-error turned into an exit of the forked stream-tests JVM.
Leave the loader open and disable jvm-exit-on-fatal-error for that system.

* Tidy zstd test idioms

Motivation:
Review nits: an[ClassNotFoundException] should read a[ClassNotFoundException],
and charset names should not be passed as strings.

Modification:
Use a[ClassNotFoundException] in CompressionWithoutZstdSpec and
StandardCharsets.UTF_8 in ZstdSpec.

Result:
No behaviour change.

Tests:
sbt "stream-tests/testOnly *ZstdSpec *CompressionWithoutZstdSpec" (24 passed)

References:
None

* Fix zstd decompression of chunk-aligned output and address review

Motivation:
ZstdDecompressor failed valid streams whose decompressed size is an exact
multiple of maxBytesPerChunk. When the output buffer filled exactly at the end
of a frame, outputPending stayed true and one more decompress call was made.
zstd then waited for a new frame and the stage failed with "Truncated zstd
stream" after all data had been emitted. The default chunk is 128 KiB, so any
128 KiB-aligned payload was affected. The review also found an incomplete
operator index, inaccurate operator docs, String dictionaries encoded with the
platform default charset, and a Java source under src/main/scala.

Modification:
- Only treat output as pending while the frame is incomplete: a completed frame
  is fully flushed.
- Add the zstd operators to the generated stream operator index.
- Document the zstd overloads and autoFlush, place Dependency before Signature,
  note that zstdDecompress does not support dictionaries, and fix the
  zstdDecompress error description.
- Encode String dictionaries as UTF-8 in the scaladsl and javadsl 
ZstdDictionary.
- Move javadsl ZstdDictionary.java to src/main/java and make its field private.
- Use imports instead of fully qualified names in new code.

Result:
Chunk-aligned and concatenated zstd streams decompress correctly, and String
dictionaries give the same bytes on every node.

Tests:
- New ZstdSpec test for chunk-aligned single and concatenated frames (fails
  without the fix with "Truncated zstd stream").
- New ZstdSpec test for UTF-8 String dictionaries.
- sbt "actor-tests/testOnly org.apache.pekko.util.ByteStringSpec" (246 passed)
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.*" (149 
passed)
- sbt "actor/mimaReportBinaryIssues" "stream/mimaReportBinaryIssues"
- scalafmt and javafmtCheck (JDK 17)

References:
Review https://github.com/apache/pekko/pull/3639#pullrequestreview-5624956465

* Add dictionary support to zstdDecompress

Motivation:
Compression.zstd accepts a dictionary, but zstdDecompress did not, so output
compressed with a dictionary could not be decompressed with Pekko Streams.

Modification:
- Add zstdDecompress(maxBytesPerChunk, dictionary) to the scaladsl and javadsl
  Compression APIs.
- ZstdDecompressor builds one ZstdDictDecompress per stage, which digests the
  dictionary (or its offset/length slice, without copying it in Java) once, and
  loads it by reference into the decompression context of every
  materialization. The dictionary level only applies to compression and is
  ignored. A dictionary with only one of offset and length is rejected when the
  flow is created.
- Document the new overload.

Result:
Data compressed with a zstd dictionary can be decompressed with zstdDecompress.

Tests:
- New ZstdSpec tests: round trips with a whole dictionary and a dictionary slice
  through the scaladsl and javadsl APIs, using incompressible data so the frames
  must reference the dictionary; decompressing without the dictionary fails; an
  invalid offset/length combination is rejected.
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.* 
org.apache.pekko.stream.scaladsl.CompressionSpec"
  (158 passed)
- sbt "stream/mimaReportBinaryIssues" "bench-jmh/compile"

References:
https://github.com/apache/pekko/pull/3639

* feat: mark the zstd compression API as ApiMayChange
Motivation:
The zstd API is new in 2.0.0 and is likely to change: limiting the
compression and decompression windows (RFC 9659) needs new parameters
once zstd-jni exposes them, and the scaladsl ZstdDictionary case class
cannot gain fields without breaking binary compatibility.

Modification:
- Mark the zstd and zstdDecompress overloads and the Zstd* constants in
  scaladsl and javadsl Compression as @ApiMayChange.
- Mark the scaladsl ZstdDictionary class and companion, and the javadsl
  ZstdDictionary class, as @ApiMayChange.
- Document toImpl as INTERNAL API in both ZstdDictionary classes, since
  the javadsl one is public to Java callers.

Result:
The zstd API can still evolve after 2.0.0.

Tests:
- Annotations and docs only, no behaviour change
- sbt "stream/compile" "stream/javafmtCheck"
- scalafmt on the changed Scala files

References:
https://github.com/apache/pekko/pull/3639

* feat: add zstd compression and decompression settings, with windowLog
Motivation:
Every new zstd option needed another Compression.zstd or zstdDecompress
overload in both DSLs. Limiting the compression window (RFC 9659 caps the
HTTP zstd content-coding at windowLog 23) would have added a fourth
compression overload, and the decompression window limit is next.

Modification:
- Add ZstdCompressionSettings (level, autoFlush, windowLog, dictionary) and
  ZstdDecompressionSettings (maxBytesPerChunk, dictionary) in
  org.apache.pekko.stream, shared by both DSLs like RestartSettings. They are
  final classes with private constructors and withX methods, so fields can
  be added compatibly. withDictionary accepts the scaladsl and the javadsl
  ZstdDictionary. Both are @ApiMayChange.
- Add zstd(settings) and zstdDecompress(settings) to both DSLs. Keep zstd,
  zstd(level, dictionary) and zstdDecompress(maxBytesPerChunk), and remove
  the unreleased zstd(level, dictionary, autoFlush) and
  zstdDecompress(maxBytesPerChunk, dictionary) overloads.
- ZstdCompressor sets ZstdCompressCtx.setWindowLog when windowLog is not 0.
  Add the ZstdMinWindowLog and ZstdMaxWindowLog constants.
- Update the zstd and zstdDecompress operator pages.

Result:
The zstd window can be limited for peers such as HTTP clients, and new zstd
options no longer need new overloads.

Tests:
- New ZstdSpec tests: a frame compressed at level 22 is rejected by a decoder
  limited to windowLog 23 and accepted with withWindowLog(23), from both
  DSLs and with a dictionary; settings defaults, validation and withX
  methods. Existing dictionary and auto-flush tests use the settings.
- sbt "stream-tests/testOnly ...ZstdSpec ...ZstdAutoFlushSpec 
...CompressionWithoutZstdSpec" (62 passed, 1 canceled by an existing assume)
- sbt "++3.3.8" "stream/compile" "stream-tests/Test/compile"
- sbt "stream/javafmtCheck", "stream/headerCreate", scalafmt on changed files

References:
https://github.com/apache/pekko/pull/3639
Refs #3674

* fix: release zstd compressor native resources on setup failure and close
Motivation:
Review of #3639 found that ZstdCompressor could leak native memory: a
failing dictionary setup (e.g. a negative slice offset) threw after the
ZstdCompressCtx was allocated, and the stage never got to close it. A
ZstdDictCompress was never closed explicitly, so its memory waited for
finalization and leaked with finalization disabled. Neither has a Cleaner.

Modification:
- Close the context if setting the level, window log or dictionary fails,
  and close a ZstdDictCompress if loading it into the context fails.
- Keep the ZstdDictCompress and close it after the context in close().
- Make the javadsl ZstdDictionary final.

Result:
The zstd compressor releases its native memory on every path.

Tests:
- New ZstdSpec tests: the context is closed after a failed setup, and both
  dictionary-with-level variants are closed with the compressor. Both fail
  without the fix.
- sbt "stream-tests/testOnly ...ZstdSpec ...ZstdAutoFlushSpec 
...CompressionWithoutZstdSpec ...CompressionSpec"
- sbt "++3.3.8" "stream-tests/Test/compile"
- sbt "stream/javafmtCheck", scalafmt on the changed Scala files

References:
https://github.com/apache/pekko/pull/3639

* fix: free the zstd decompression dictionary with its context
Motivation:
ZstdDecompressor digested the dictionary once into a ZstdDictDecompress
shared by every materialization. It could not be closed when one
materialization stopped and has no Cleaner, so its native memory was only
freed by finalization, and never with finalization disabled. This is the
decompression side of the ZstdDictCompress review finding on #3639.

Modification:
- Load the dictionary bytes into each materialization's ZstdDecompressCtx,
  which owns the digested dictionary and frees it when the stage closes
  the context.
- Copy the dictionary bytes, or the slice, when the operator is created, and
  validate the slice bounds there.
- Close the context if loading the dictionary fails, since postStop is not
  called then.

Result:
The zstd decompressor no longer relies on finalization to free dictionary
memory. Each materialization digests the dictionary itself, as the
compressor already does for dictionaries without a level.

Tests:
- New ZstdSpec tests: slices outside the dictionary are rejected when the
  operator is created, and the operator decompresses with a copy of the
  dictionary, for more than one materialization. The native memory itself
  is not observable from the tests.
- sbt "stream-tests/testOnly ...ZstdSpec ...ZstdAutoFlushSpec 
...CompressionWithoutZstdSpec ...CompressionSpec" (73 passed, 1 canceled by an 
existing assume)
- scalafmt on the changed files

References:
https://github.com/apache/pekko/pull/3639

* fix: validate zstd compression dictionary slices and allow them without a 
level

Motivation:
Review of #3639: ZstdCompressor did not check that a dictionary slice is
within the dictionary, so a bad offset or length failed in zstd-jni with
a less clear error. A slice without a level, which the Java
ZstdDictionary constructors allow and zstdDecompress accepts, was
rejected by the compressor.

Modification:
- Share the slice bounds check between ZstdCompressor and
  ZstdDecompressor in the internal ZstdDictionaryImpl object.
- ZstdCompressor checks the slice before digesting it, and loads a copy
  of the slice into the context when no level is set.

Result:
Out-of-bounds dictionary slices fail with the same
IllegalArgumentException for compression and decompression, and a slice
without a level can be used for compression.

Tests:
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.ZstdSpec 
org.apache.pekko.stream.io.compression.ZstdAutoFlushSpec 
org.apache.pekko.stream.io.compression.CompressionWithoutZstdSpec": 76 passed
- With the ZstdCompressor change reverted, the new slice tests fail

References:
Refs #3639

* fix: hold the scaladsl ZstdDictionary bytes in a ByteString

Motivation:
Review of #3639: the scaladsl ZstdDictionary is a case class with an
Array[Byte] field, so it was mutable and its equals and hashCode
compared the array by reference.

Modification:
- The dictionary field is now an immutable ByteString. The Array[Byte]
  factory methods copy the array.
- toImpl hands the operators a copy of the bytes, so whoever holds the
  impl cannot modify the dictionary.

Result:
The scaladsl ZstdDictionary is immutable and has value semantics.

Tests:
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.ZstdSpec 
org.apache.pekko.stream.io.compression.ZstdAutoFlushSpec 
org.apache.pekko.stream.io.compression.CompressionWithoutZstdSpec 
org.apache.pekko.stream.scaladsl.CompressionSpec": 81 passed
- sbt "++3.3" "stream-tests/Test/compile": ok

References:
Refs #3639

* fix: make the javadsl ZstdDictionary immutable

Motivation:
The javadsl ZstdDictionary had setters and held the caller's byte[],
so it could be changed after an operator was created from it, and it
compared by reference. The scaladsl ZstdDictionary was already made
immutable.

Modification:
- Final fields, with the bytes held in a ByteString. The byte[]
  constructors copy the array.
- Remove the setters. getDictionary returns the ByteString.
- Add equals, hashCode and toString.
- toImpl hands the operators a copy of the bytes.

Result:
The javadsl ZstdDictionary is immutable and has value semantics, the
same as the scaladsl one.

Tests:
- sbt "stream/javafmt" (JDK 17)
- sbt "stream-tests/testOnly org.apache.pekko.stream.io.compression.ZstdSpec 
org.apache.pekko.stream.io.compression.ZstdAutoFlushSpec 
org.apache.pekko.stream.io.compression.CompressionWithoutZstdSpec 
org.apache.pekko.stream.scaladsl.CompressionSpec": 83 passed
- sbt "++3.3" "stream-tests/Test/compile": ok

References:
Refs #3639

---------

Co-authored-by: Matthew de Detrich <[email protected]>

Report URL: https://github.com/apache/pekko/actions/runs/40025866651

With regards,
GitHub Actions via GitBox


---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]

Reply via email to