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]
