diff --git a/conf/defaults.yaml b/conf/defaults.yaml index 4368099725c..a64bd5b8940 100644 --- a/conf/defaults.yaml +++ b/conf/defaults.yaml @@ -54,7 +54,10 @@ storm.nimbus.zookeeper.acls.fixup: true storm.auth.simple-white-list.users: [] storm.cluster.state.store: "org.apache.storm.cluster.ZKStateStorageFactory" -storm.meta.serialization.delegate: "org.apache.storm.serialization.GzipThriftSerializationDelegate" +storm.meta.serialization.delegate: "org.apache.storm.serialization.ZstdBridgeThriftSerializationDelegate" +storm.compression.zstd.level: 3 +storm.compression.zstd.max.decompressed.bytes: 104857600 +storm.compression.gzip.max.decompressed.bytes: 104857600 storm.codedistributor.class: "org.apache.storm.codedistributor.LocalFileSystemCodeDistributor" storm.workers.artifacts.dir: "workers-artifacts" storm.health.check.dir: "healthchecks" diff --git a/docs/Cluster-State-Serialization.md b/docs/Cluster-State-Serialization.md new file mode 100644 index 00000000000..8581448f409 --- /dev/null +++ b/docs/Cluster-State-Serialization.md @@ -0,0 +1,124 @@ +--- +title: Cluster State Serialization +layout: documentation +documentation: true +--- + +This page describes how Storm serializes the *meta* state it persists in +ZooKeeper (and other configured state stores) such as topology assignments, Nimbus +summaries, `StormBase` records, log configs, credentials, worker heartbeats, +profile requests, errors, etc. + +It is distinct from +[tuple serialization](Serialization.html), which covers payloads exchanged +between spouts and bolts at runtime via Kryo. + +## Background + +All cluster state writes go through `Utils.serialize(...)` / +`Utils.deserialize(...)`, which in turn delegate to a pluggable +`SerializationDelegate` selected by the +`storm.meta.serialization.delegate` config. + +## Configuration + +| Key | Default | Range | Description | +|---|---|---|---| +| `storm.meta.serialization.delegate` | `org.apache.storm.serialization.ZstdBridgeThriftSerializationDelegate` | any `SerializationDelegate` impl | Class used to (de)serialize cluster state. | +| `storm.compression.zstd.level` | `3` | `1`–`19` | Zstandard compression level. Higher = smaller + slower. Levels 20–22 are rejected by the validator. | +| `storm.compression.zstd.max.decompressed.bytes` | `104857600` (100 MiB) | `> 0` | Hard cap on the size of any zstd-decompressed payload. | +| `storm.compression.gzip.max.decompressed.bytes` | `104857600` (100 MiB) | `> 0` | Hard cap on the size of any gzip-decompressed payload. Also enforced by `GzipSerializationDelegate`. | + +## Choosing a delegate + +* **`ZstdBridgeThriftSerializationDelegate`** *(default)* — recommended. + Writes zstd, reads anything previously written. Use this unless you + have a specific reason not to. +* **`ZstdThriftSerializationDelegate`** — pure zstd, refuses non-zstd + input. Only safe to deploy after every znode in your state store has + been rewritten by a bridge delegate (e.g. by submitting / killing each + topology, or by force-rewriting Nimbus state). Use only when you want + to *enforce* the new format. +* **`GzipBridgeThriftSerializationDelegate`** — legacy default; still + available for clusters that want to roll forward without touching the + codec. +* **`ThriftSerializationDelegate`** — raw Thrift. + +## Migration to Zstandard compression + +Starting with Apache Storm 3.X, Zstandard is supported as the default +compression codec for cluster state, replacing gzip for better +performance — faster compression and decompression at comparable or +better ratios. Earlier versions used `GzipThriftSerializationDelegate`, +wrapped by `GzipBridgeThriftSerializationDelegate` to allow rolling +upgrades from clusters that had previously stored raw Thrift bytes; the +new `ZstdBridgeThriftSerializationDelegate` plays the equivalent bridge +role for the gzip to zstd transition. + +| Area | Gzip | Zstandard | +|---|---------------------------------------------------------|-----------------------------------------------------| +| Default delegate | `GzipThriftSerializationDelegate` (via `GzipBridge...`) | `ZstdBridgeThriftSerializationDelegate` | +| Compression codec | gzip (`java.util.zip`) | Zstandard (via `commons-compress` + `zstd-jni`) | +| Decompression bound | none | bounded (`BoundedInputStream`), default 100 MiB | +| Format detection | gzip magic only | gzip magic *and* zstd magic | +| Config validation | none for compression | `ZstdLevelValidator` (1–19), positive bounds checks | + +### Zstandard `SerializationDelegate` implementations + +* `ZstdThriftSerializationDelegate`: pure zstd Thrift codec. Serializes + any `TBase` with zstd at the configured level; deserialization + requires the input to begin with the zstd magic number + (`0xFD2FB528`). +* `ZstdBridgeThriftSerializationDelegate`: the new default, implemented to + allow rolling upgrades from clusters that had previously stored payloads + as gzip-compressed. Always *writes* zstd. On read, dispatches based on a + magic-byte sniff: + +``` +ZstdBridgeThriftSerializationDelegate.deserialize(bytes) + ├── bytes starts with zstd magic (0xFD2FB528) delegates to ZstdThriftSerializationDelegate + └── otherwise, delegates to GzipBridgeThriftSerializationDelegate.deserialize(bytes) + ├── bytes starts with gzip magic (0x1F8B) delegates to GzipThriftSerializationDelegate + └── otherwise delegates to ThriftSerializationDelegate (raw Thrift) +``` + +This delegation chain is the key property that makes the new default +rolling-upgrade safe: nodes running the new code can still read every +older payload that may already exist in ZooKeeper, while new writes use +zstd. + +### Zip-bomb protection + +`GzipUtils.decompress` and `ZstdUtils.decompress` (both in +`org.apache.storm.utils.Utils`) wrap the decompressor stream in an Apache +Commons `BoundedInputStream` with `maxCount` set to the configured cap. +After draining the bounded stream, the underlying decompressor is probed +with one extra `read()`; if any byte remains, the call fails with: + +``` +Decompression threshold exceeded! Possible security risk or invalid data size. +``` + +The same guard is applied to the legacy `GzipSerializationDelegate` (the +non-Thrift Java-serialization variant). + +### Upgrading an existing cluster + +1. **Roll Nimbus and Supervisors onto the new build.** The bridge + delegate is the default, so no config change is required for a safe + upgrade. +2. **(Optional) Tune `storm.compression.zstd.level`** if you want a + tighter compression / latency trade-off. Most state writes are + infrequent; level 3 is a good default. +3. **(Optional) Tune `storm.compression.zstd.max.decompressed.bytes`** if + you legitimately persist payloads larger than 100 MiB. The cap + guards against malformed or hostile data, raise it deliberately. +4. **(Optional) Switch to the strict `ZstdThriftSerializationDelegate`** + *only* after every legacy payload has been rewritten. The bridge + delegate is sufficient for the vast majority of deployments. + +### Dependencies + +The zstd codec is provided by Apache Commons Compress +(`org.apache.commons:commons-compress`) backed by the `com.github.luben:zstd-jni` +native binding. diff --git a/docs/Serialization.md b/docs/Serialization.md index 4d2749105a6..0e7cc2e87a3 100644 --- a/docs/Serialization.md +++ b/docs/Serialization.md @@ -5,6 +5,8 @@ documentation: true --- This page is about how the serialization system in Storm works for versions 0.6.0 and onwards. Storm used a different serialization system prior to 0.6.0 which is documented on [Serialization (prior to 0.6.0)](Serialization-\(prior-to-0.6.0\).html). +> This page covers **tuple** serialization (data flowing between spouts and bolts). For how Storm serializes the meta state it persists in ZooKeeper and related configuration, see [Cluster State Serialization](Cluster-State-Serialization.html). + Tuples can be comprised of objects of any types. Since Storm is a distributed system, it needs to know how to serialize and deserialize objects when they're passed between tasks. Storm uses [Kryo](https://github.com/EsotericSoftware/kryo) for serialization. Kryo is a flexible and fast serialization library that produces small serializations. diff --git a/pom.xml b/pom.xml index 4d3863693ac..c1ba6755507 100644 --- a/pom.xml +++ b/pom.xml @@ -82,6 +82,7 @@ 1.28.0 + 1.5.7-8 2.22.0 3.20.0 1.6.0 @@ -514,6 +515,11 @@ commons-compress ${commons-compress.version} + + com.github.luben + zstd-jni + ${zstd-jni.version} + org.apache.commons commons-exec diff --git a/storm-client/pom.xml b/storm-client/pom.xml index 62d6c20d9d3..3a7ee315d1a 100644 --- a/storm-client/pom.xml +++ b/storm-client/pom.xml @@ -135,6 +135,14 @@ curator-test test + + org.apache.commons + commons-compress + + + com.github.luben + zstd-jni + diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index e332b726b28..f91d30ca3c0 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -1458,7 +1458,24 @@ public class Config extends HashMap { */ @IsString public static final String STORM_META_SERIALIZATION_DELEGATE = "storm.meta.serialization.delegate"; - + /** + * GZIP max decompression bytes. Defaults to 104857600 (100MB). + */ + @IsPositiveNumber(includeZero = false) + public static final String STORM_COMPRESSION_GZIP_MAX_DECOMPRESSED_BYTES = "storm.compression.gzip.max.decompressed.bytes"; + /** + * Zstandard compression level. + * Supported range: 1 to 19. Default: 3. + * Prohibited: Levels 20-22 (Ultra mode) are not allowed as they + * require dramatically more working memory per call. + */ + @CustomValidator(validatorClass = ConfigValidation.ZstdLevelValidator.class) + public static final String STORM_COMPRESSION_ZSTD_LEVEL = "storm.compression.zstd.level"; + /** + * Zstandard max decompression bytes. Defaults to 104857600 (100MB). + */ + @IsPositiveNumber(includeZero = false) + public static final String STORM_COMPRESSION_ZSTD_MAX_DECOMPRESSED_BYTES = "storm.compression.zstd.max.decompressed.bytes"; /** * Configure the topology metrics reporters to be used on workers. */ diff --git a/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegate.java index bc966112426..67ae15ada00 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegate.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegate.java @@ -1,4 +1,4 @@ -/** +/* * 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 diff --git a/storm-client/src/jvm/org/apache/storm/serialization/GzipSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/GzipSerializationDelegate.java index 9c4045854d5..d12c2b01879 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/GzipSerializationDelegate.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/GzipSerializationDelegate.java @@ -1,4 +1,4 @@ -/** +/* * 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 @@ -20,15 +20,22 @@ import java.util.Map; import java.util.zip.GZIPInputStream; import java.util.zip.GZIPOutputStream; +import org.apache.storm.Config; +import org.apache.storm.shade.org.apache.commons.io.input.BoundedInputStream; +import org.apache.storm.utils.ObjectReader; /** * Note, this assumes it's deserializing a gzip byte stream, and will err if it encounters any other serialization. */ public class GzipSerializationDelegate implements SerializationDelegate { + private static final int DEFAULT_MAX_DECOMPRESSED_BYTES = 100 * 1024 * 1024; + private int maxDecompressedBytes; + @Override public void prepare(Map topoConf) { - // No-op + this.maxDecompressedBytes = ObjectReader.getInt(topoConf.getOrDefault(Config.STORM_COMPRESSION_GZIP_MAX_DECOMPRESSED_BYTES, + DEFAULT_MAX_DECOMPRESSED_BYTES)); } @Override @@ -47,17 +54,21 @@ public byte[] serialize(Object object) { @Override public T deserialize(byte[] bytes, Class clazz) { - try { - ByteArrayInputStream bis = new ByteArrayInputStream(bytes); - GZIPInputStream gis = new GZIPInputStream(bis); - ObjectInputStream ois = new ObjectInputStream(gis); + try (ByteArrayInputStream bis = new ByteArrayInputStream(bytes); + GZIPInputStream gis = new GZIPInputStream(bis); + BoundedInputStream lis = BoundedInputStream.builder() + .setMaxCount(this.maxDecompressedBytes) + .setInputStream(gis) + .setPropagateClose(true) + .get(); + ObjectInputStream ois = new ObjectInputStream(lis)) { Object ret = ois.readObject(); - ois.close(); - return (T) ret; - } catch (IOException ioe) { - throw new RuntimeException(ioe); - } catch (ClassNotFoundException e) { - throw new RuntimeException(e); + if (gis.read() != -1) { + throw new IOException("Decompression threshold exceeded! Possible security risk or invalid data size."); + } + return clazz.cast(ret); + } catch (IOException | ClassNotFoundException e) { + throw new RuntimeException("Deserialization failed: " + e.getMessage(), e); } } } diff --git a/storm-client/src/jvm/org/apache/storm/serialization/GzipThriftSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/GzipThriftSerializationDelegate.java index f518628a82d..ae3bc7ac1cf 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/GzipThriftSerializationDelegate.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/GzipThriftSerializationDelegate.java @@ -19,10 +19,12 @@ package org.apache.storm.serialization; import java.util.Map; +import org.apache.storm.Config; import org.apache.storm.thrift.TBase; import org.apache.storm.thrift.TDeserializer; import org.apache.storm.thrift.TException; import org.apache.storm.thrift.TSerializer; +import org.apache.storm.utils.ObjectReader; import org.apache.storm.utils.Utils; /** @@ -30,15 +32,19 @@ */ public class GzipThriftSerializationDelegate implements SerializationDelegate { + private static final int DEFAULT_MAX_DECOMPRESSED_BYTES = 100 * 1024 * 1024; + private int maxDecompressedBytes; + @Override public void prepare(Map topoConf) { - // No-op + this.maxDecompressedBytes = ObjectReader.getInt(topoConf.getOrDefault(Config.STORM_COMPRESSION_GZIP_MAX_DECOMPRESSED_BYTES, + DEFAULT_MAX_DECOMPRESSED_BYTES)); } @Override public byte[] serialize(Object object) { try { - return Utils.gzip(new TSerializer().serialize((TBase) object)); + return Utils.GzipUtils.compress(new TSerializer().serialize((TBase) object)); } catch (TException e) { throw new RuntimeException(e); } @@ -48,7 +54,7 @@ public byte[] serialize(Object object) { public T deserialize(byte[] bytes, Class clazz) { try { TBase instance = (TBase) clazz.newInstance(); - new TDeserializer().deserialize(instance, Utils.gunzip(bytes)); + new TDeserializer().deserialize(instance, Utils.GzipUtils.decompress(bytes, this.maxDecompressedBytes)); return (T) instance; } catch (Exception e) { throw new RuntimeException(e); diff --git a/storm-client/src/jvm/org/apache/storm/serialization/ZstdBridgeThriftSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/ZstdBridgeThriftSerializationDelegate.java new file mode 100644 index 00000000000..a3484b7fd17 --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/serialization/ZstdBridgeThriftSerializationDelegate.java @@ -0,0 +1,50 @@ +/* + * 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.storm.serialization; + +import java.util.Map; +import org.apache.storm.utils.Utils; + +/** + * Always writes Zstd out, but tests incoming bytes to determine the format. + * If Zstd magic is found, it uses {@link ZstdThriftSerializationDelegate}. + * If not, it falls back to {@link ThriftSerializationDelegate} for raw Thrift. + */ +public class ZstdBridgeThriftSerializationDelegate implements SerializationDelegate { + + private final GzipBridgeThriftSerializationDelegate defaultDelegate = new GzipBridgeThriftSerializationDelegate(); + private final ZstdThriftSerializationDelegate zstdDelegate = new ZstdThriftSerializationDelegate(); + + @Override + public void prepare(Map topoConf) { + defaultDelegate.prepare(topoConf); + zstdDelegate.prepare(topoConf); + } + + @Override + public byte[] serialize(Object object) { + // Always compress new data with Zstd + return zstdDelegate.serialize(object); + } + + @Override + public T deserialize(byte[] bytes, Class clazz) { + if (Utils.ZstdUtils.isZstd(bytes)) { + return zstdDelegate.deserialize(bytes, clazz); + } else { + // Fallback to ZstdBridgeThriftSerializationDelegate + // it delegates to the proper SerializationDelegate (GzipThriftSerializationDelegate or ThriftSerializationDelegate) + return defaultDelegate.deserialize(bytes, clazz); + } + } +} diff --git a/storm-client/src/jvm/org/apache/storm/serialization/ZstdThriftSerializationDelegate.java b/storm-client/src/jvm/org/apache/storm/serialization/ZstdThriftSerializationDelegate.java new file mode 100644 index 00000000000..d3b990a7b3e --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/serialization/ZstdThriftSerializationDelegate.java @@ -0,0 +1,87 @@ +/* + * 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.storm.serialization; + +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.thrift.TBase; +import org.apache.storm.thrift.TDeserializer; +import org.apache.storm.thrift.TException; +import org.apache.storm.thrift.TSerializer; +import org.apache.storm.thrift.transport.TTransportException; +import org.apache.storm.utils.ObjectReader; +import org.apache.storm.utils.Utils; + +/** + * Note, this assumes it's deserializing a zstd byte stream, and will err if it encounters any other serialization. + */ +public class ZstdThriftSerializationDelegate implements SerializationDelegate { + + private static final int DEFAULT_MAX_DECOMPRESSED_BYTES = 100 * 1024 * 1024; + private static final int DEFAULT_ZSTD_COMPRESSION_LEVEL = 3; + + private int zstdCompressionLevel; + private int maxDecompressedBytes; + + @Override + public void prepare(Map topoConf) { + this.zstdCompressionLevel = ObjectReader.getInt(topoConf.getOrDefault(Config.STORM_COMPRESSION_ZSTD_LEVEL, + DEFAULT_ZSTD_COMPRESSION_LEVEL)); + this.maxDecompressedBytes = ObjectReader.getInt(topoConf.getOrDefault(Config.STORM_COMPRESSION_ZSTD_MAX_DECOMPRESSED_BYTES, + DEFAULT_MAX_DECOMPRESSED_BYTES)); + } + + @Override + public byte[] serialize(Object object) { + if (!(object instanceof TBase)) { + throw new IllegalArgumentException("Object must be an instance of TBase"); + } + try { + TSerializer serializer = new TSerializer(); + byte[] thriftData = serializer.serialize((TBase) object); + return Utils.ZstdUtils.compress(thriftData, this.zstdCompressionLevel); + } catch (TTransportException e) { + throw new RuntimeException("Failed to initialize Thrift Serializer", e); + } catch (TException e) { + throw new RuntimeException("Failed to serialize Thrift object", e); + } + } + + @Override + public T deserialize(byte[] bytes, Class clazz) { + if (!Utils.ZstdUtils.isZstd(bytes)) { + throw new RuntimeException( + String.format("Cannot deserialize [%s]. Expected zstd compressed bytes, but received unknown format.", + clazz.getSimpleName()) + ); + } + try { + TDeserializer deserializer = new TDeserializer(); + byte[] decompressed = Utils.ZstdUtils.decompress(bytes, this.maxDecompressedBytes); + TBase instance = clazz.asSubclass(TBase.class).getDeclaredConstructor().newInstance(); + deserializer.deserialize(instance, decompressed); + return (T) instance; + } catch (TTransportException e) { + throw new RuntimeException("Failed to initialize Thrift Deserializer", e); + } catch (ReflectiveOperationException | TException e) { + throw new RuntimeException("Failed to deserialize bytes to " + clazz.getName(), e); + } + } + +} diff --git a/storm-client/src/jvm/org/apache/storm/utils/Utils.java b/storm-client/src/jvm/org/apache/storm/utils/Utils.java index b639863f836..7ba3298fee2 100644 --- a/storm-client/src/jvm/org/apache/storm/utils/Utils.java +++ b/storm-client/src/jvm/org/apache/storm/utils/Utils.java @@ -34,6 +34,8 @@ import java.io.Serializable; import java.io.UnsupportedEncodingException; import java.lang.Thread.UncaughtExceptionHandler; +import java.lang.invoke.MethodHandles; +import java.lang.invoke.VarHandle; import java.lang.management.ManagementFactory; import java.lang.management.ThreadInfo; import java.net.InetAddress; @@ -43,6 +45,7 @@ import java.net.URLEncoder; import java.net.UnknownHostException; import java.nio.ByteBuffer; +import java.nio.ByteOrder; import java.nio.charset.StandardCharsets; import java.nio.file.Files; import java.util.ArrayList; @@ -72,6 +75,9 @@ import java.util.zip.ZipEntry; import java.util.zip.ZipFile; import javax.security.auth.Subject; +import org.apache.commons.compress.compressors.zstandard.ZstdCompressorInputStream; +import org.apache.commons.compress.compressors.zstandard.ZstdCompressorOutputStream; +import org.apache.commons.io.IOUtils; import org.apache.storm.Config; import org.apache.storm.blobstore.BlobStore; import org.apache.storm.blobstore.ClientBlobStore; @@ -96,6 +102,7 @@ import org.apache.storm.shade.net.minidev.json.JSONValue; import org.apache.storm.shade.net.minidev.json.parser.ParseException; import org.apache.storm.shade.org.apache.commons.io.FileUtils; +import org.apache.storm.shade.org.apache.commons.io.input.BoundedInputStream; import org.apache.storm.shade.org.apache.commons.io.input.ClassLoaderObjectInputStream; import org.apache.storm.shade.org.apache.commons.lang3.StringUtils; import org.apache.storm.shade.org.apache.zookeeper.ZooDefs; @@ -930,33 +937,181 @@ public static List tuple(Object... values) { return ret; } - public static byte[] gzip(byte[] data) { - try { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - GZIPOutputStream out = new GZIPOutputStream(bos); - out.write(data); - out.close(); - return bos.toByteArray(); - } catch (IOException e) { - throw new RuntimeException(e); + /** + * Static utility class for GZIP compression and decompression. + */ + public static final class GzipUtils { + + // GZIP magic number (first two bytes): 0x1F8B. + private static final byte GZIP_MAGIC_0 = (byte) 0x1F; + private static final byte GZIP_MAGIC_1 = (byte) 0x8B; + + private static final int BUFFER_SIZE = 64 * 1024; + + /** + * Private constructor to prevent instantiation. + * @throws UnsupportedOperationException if an attempt is made to instantiate this class. + */ + private GzipUtils() { + throw new UnsupportedOperationException("Utility class should not be instantiated."); + } + + /** + * Compresses the provided byte array using GZIP. + * + * @param data the raw byte array to compress. + * @return a compressed byte array, or the original array if null/empty. + * @throws RuntimeException wrapping an {@link IOException} if the compression fails. + */ + public static byte[] compress(byte[] data) { + if (data == null || data.length == 0) { + return data; + } + try (ByteArrayOutputStream bos = new ByteArrayOutputStream(data.length); + GZIPOutputStream gzipOut = new GZIPOutputStream(bos, BUFFER_SIZE)) { + gzipOut.write(data); + gzipOut.finish(); + return bos.toByteArray(); + } catch (IOException e) { + throw new RuntimeException("GZIP compression failed", e); + } + } + + /** + * Decompresses a GZIP-compressed byte array. + * + * @param data the compressed byte array (GZIP frame). + * @param maxDecompressedBytes maximum number of bytes allowed after decompression, + * as a safeguard against zip-bomb attacks. + * @return the original decompressed byte array, or the input if null/empty. + * @throws RuntimeException wrapping an {@link IOException} if the decompression fails, + * if the data is not a valid GZIP stream, or if the decompressed + * size exceeds {@code maxDecompressedBytes}. + */ + public static byte[] decompress(byte[] data, int maxDecompressedBytes) { + if (data == null || data.length == 0) { + return data; + } + try (ByteArrayInputStream bis = new ByteArrayInputStream(data); + GZIPInputStream gzipIn = new GZIPInputStream(bis, BUFFER_SIZE); + BoundedInputStream limitedIn = BoundedInputStream.builder() + .setInputStream(gzipIn) + .setMaxCount(maxDecompressedBytes) + .get(); + ByteArrayOutputStream bos = new ByteArrayOutputStream()) { + IOUtils.copy(limitedIn, bos); + if (gzipIn.read() != -1) { + throw new IOException("Decompression threshold exceeded! Possible security risk or invalid data size."); + } + return bos.toByteArray(); + } catch (IOException e) { + throw new RuntimeException("GZIP decompression failed", e); + } + } + + /** + * Checks the first 2 bytes of the array against the GZIP magic number (0x1F8B). + * + * @param bytes the data payload. + * @return {@code true} if the payload starts with a valid GZIP header, {@code false} otherwise. + */ + public static boolean isGzip(byte[] bytes) { + if (bytes == null || bytes.length < 2) { + return false; + } + return bytes[0] == GZIP_MAGIC_0 && bytes[1] == GZIP_MAGIC_1; } } - public static byte[] gunzip(byte[] data) { - try { - ByteArrayOutputStream bos = new ByteArrayOutputStream(); - ByteArrayInputStream bis = new ByteArrayInputStream(data); - GZIPInputStream in = new GZIPInputStream(bis); - byte[] buffer = new byte[1024]; - int len = 0; - while ((len = in.read(buffer)) >= 0) { - bos.write(buffer, 0, len); + /** + * Static utility class for Zstandard (Zstd) compression and decompression. + */ + public static final class ZstdUtils { + + /** + * Zstandard magic number 0xFD2FB528. + */ + private static final int ZSTD_MAGIC_INT = 0xFD2FB528; + private static final VarHandle INT_HANDLE = MethodHandles.byteArrayViewVarHandle(int[].class, ByteOrder.LITTLE_ENDIAN); + private static final int BUFFER_SIZE = 64 * 1024; + + /** + * Private constructor to prevent instantiation. + * @throws UnsupportedOperationException if an attempt is made to instantiate this class. + */ + private ZstdUtils() { + throw new UnsupportedOperationException("Utility class should not be instantiated."); + } + + /** + * Compresses the provided byte array using Zstandard. + * + *

The output includes the standard Zstandard frame header, making it + * self-describing for the decompression phase.

+ * + * @param data the raw byte array to compress. + * @param compressionLevel the zstd compression level. + * @return a compressed byte array, or the original array if null/empty. + * @throws RuntimeException wrapping an {@link IOException} if the compression fails. + */ + public static byte[] compress(byte[] data, int compressionLevel) { + if (data == null || data.length == 0) { + return data; } - in.close(); - bos.close(); - return bos.toByteArray(); - } catch (IOException e) { - throw new RuntimeException(e); + + try (ByteArrayOutputStream bos = new ByteArrayOutputStream(data.length)) { + try (ZstdCompressorOutputStream zstdOut = ZstdCompressorOutputStream.builder() + .setOutputStream(bos) + .setBufferSize(BUFFER_SIZE) // impacts on compression ratio + .setLevel(compressionLevel) + .get()) { + zstdOut.write(data); + zstdOut.finish(); + } + return bos.toByteArray(); + } catch (IOException e) { + throw new RuntimeException("Zstd compression failed", e); + } + } + + /** + * Decompresses a Zstandard-compressed byte array. + * + * @param data the compressed byte array (Zstd frame). + * @return the original decompressed byte array, or the input if null/empty. + * @throws RuntimeException wrapping an {@link IOException} if the decompression fails + * or if the data is not a valid Zstd frame. + */ + public static byte[] decompress(byte[] data, int maxDecompressedBytes) { + if (data == null || data.length == 0) { + return data; + } + + try (ByteArrayInputStream bis = new ByteArrayInputStream(data); + ZstdCompressorInputStream zstdIn = new ZstdCompressorInputStream(bis); + BoundedInputStream limitedIn = BoundedInputStream.builder().setInputStream(zstdIn).setMaxCount(maxDecompressedBytes).get(); + ByteArrayOutputStream bos = new ByteArrayOutputStream()) { + IOUtils.copy(limitedIn, bos); + if (zstdIn.read() != -1) { + throw new IOException("Decompression threshold exceeded! Possible security risk or invalid data size."); + } + return bos.toByteArray(); + } catch (IOException e) { + throw new RuntimeException("Zstd decompression failed", e); + } + } + + /** + * Checks the first 4 bytes of the array against the Zstd Magic Number. + * + * @param bytes The data payload + * @return true if the payload contains a valid zstd header, false otherwise. + */ + public static boolean isZstd(byte[] bytes) { + if (bytes == null || bytes.length < 4) { + return false; + } + return (int) INT_HANDLE.get(bytes, 0) == ZSTD_MAGIC_INT; } } diff --git a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java index 4175ee9f4a7..64f0fec9523 100644 --- a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java +++ b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java @@ -803,6 +803,26 @@ public void validateField(String name, Object o) { } } + public static class ZstdLevelValidator extends Validator { + private static final int MIN_LEVEL = 1; + private static final int MAX_LEVEL = 19; + + @Override + public void validateField(String name, Object o) { + if (o == null) { + return; + } + SimpleTypeValidator.validateField(name, Integer.class, o); + int level = (Integer) o; + if (level < MIN_LEVEL || level > MAX_LEVEL) { + throw new IllegalArgumentException( + String.format("Field '%s' is invalid: %d. Zstd compression level must be between %d and %d.", + name, level, MIN_LEVEL, MAX_LEVEL) + ); + } + } + } + public static class EventLoggerRegistryValidator extends Validator { @Override diff --git a/storm-client/test/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegateTest.java b/storm-client/test/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegateTest.java index 7a5634dcc24..47c9147d54c 100644 --- a/storm-client/test/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegateTest.java +++ b/storm-client/test/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegateTest.java @@ -12,6 +12,7 @@ package org.apache.storm.serialization; +import java.util.Collections; import org.apache.storm.generated.GlobalStreamId; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -25,6 +26,7 @@ public class GzipBridgeThriftSerializationDelegateTest { @BeforeEach public void setUp() throws Exception { testDelegate = new GzipBridgeThriftSerializationDelegate(); + testDelegate.prepare(Collections.emptyMap()); } @Test diff --git a/storm-client/test/jvm/org/apache/storm/serialization/ZstdBridgeThriftSerializationDelegateTest.java b/storm-client/test/jvm/org/apache/storm/serialization/ZstdBridgeThriftSerializationDelegateTest.java new file mode 100644 index 00000000000..13bf7515cc8 --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/serialization/ZstdBridgeThriftSerializationDelegateTest.java @@ -0,0 +1,245 @@ +/* + * 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.storm.serialization; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mockStatic; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.utils.Utils; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.MockedStatic; +import org.mockito.junit.jupiter.MockitoExtension; + + +@ExtendWith(MockitoExtension.class) +class ZstdBridgeThriftSerializationDelegateTest { + + @Mock + private GzipBridgeThriftSerializationDelegate defaultDelegate; + + @Mock + private ZstdThriftSerializationDelegate zstdDelegate; + + private ZstdBridgeThriftSerializationDelegate delegate; + + private static final Map TOPO_CONF = Collections.emptyMap(); + + private static final byte[] ZSTD_BYTES = {(byte) 0x28, (byte) 0xB5, (byte) 0x2F, (byte) 0xFD, 0x00}; + private static final byte[] PLAIN_BYTES = {0x00, 0x01, 0x02, 0x03, 0x04}; + private static final byte[] RESULT_BYTES = {(byte) 0xAA, (byte) 0xBB}; + + + @BeforeEach + void setUp() throws Exception { + delegate = new ZstdBridgeThriftSerializationDelegate(); + + Field defaultField = ZstdBridgeThriftSerializationDelegate.class.getDeclaredField("defaultDelegate"); + defaultField.setAccessible(true); + defaultField.set(delegate, defaultDelegate); + + Field zstdField = ZstdBridgeThriftSerializationDelegate.class.getDeclaredField("zstdDelegate"); + zstdField.setAccessible(true); + zstdField.set(delegate, zstdDelegate); + } + + @Test + void prepare_delegatesToBothDelegates() { + Map conf = new HashMap<>(); + conf.put("key", "value"); + + // user defined conf + conf.put(Config.STORM_COMPRESSION_ZSTD_LEVEL, 3); + conf.put(Config.STORM_COMPRESSION_ZSTD_MAX_DECOMPRESSED_BYTES, 2 * 1024 * 1024); + conf.put(Config.STORM_COMPRESSION_GZIP_MAX_DECOMPRESSED_BYTES, 2 * 1024 * 1024); + + delegate.prepare(conf); + + verify(defaultDelegate).prepare(conf); + verify(zstdDelegate).prepare(conf); + verifyNoMoreInteractions(defaultDelegate, zstdDelegate); + } + + @Test + void prepare_emptyConf_doesNotThrow() { + assertDoesNotThrow(() -> delegate.prepare(TOPO_CONF)); + verify(defaultDelegate).prepare(TOPO_CONF); + verify(zstdDelegate).prepare(TOPO_CONF); + } + + @Test + void prepare_nullConf_propagatesToBothDelegates() { + delegate.prepare(null); + verify(defaultDelegate).prepare(null); + verify(zstdDelegate).prepare(null); + } + + @Test + void serialize_alwaysUsesZstdDelegate() { + Object payload = new Object(); + when(zstdDelegate.serialize(payload)).thenReturn(RESULT_BYTES); + + byte[] result = delegate.serialize(payload); + + assertArrayEquals(RESULT_BYTES, result); + verify(zstdDelegate).serialize(payload); + verifyNoInteractions(defaultDelegate); + } + + @Test + void serialize_nullObject_delegatedToZstd() { + when(zstdDelegate.serialize(null)).thenReturn(RESULT_BYTES); + + byte[] result = delegate.serialize(null); + + assertArrayEquals(RESULT_BYTES, result); + verify(zstdDelegate).serialize(null); + verifyNoInteractions(defaultDelegate); + } + + @Test + void serialize_neverUsesDefaultDelegate() { + when(zstdDelegate.serialize(any())).thenReturn(RESULT_BYTES); + + delegate.serialize("anything"); + + verifyNoInteractions(defaultDelegate); + } + + @Test + void deserialize_zstdMagic_usesZstdDelegate() { + try (MockedStatic mocked = mockStatic(Utils.ZstdUtils.class)) { + mocked.when(() -> Utils.ZstdUtils.isZstd(ZSTD_BYTES)).thenReturn(true); + when(zstdDelegate.deserialize(ZSTD_BYTES, String.class)).thenReturn("zstd-result"); + + String result = delegate.deserialize(ZSTD_BYTES, String.class); + + assertEquals("zstd-result", result); + verify(zstdDelegate).deserialize(ZSTD_BYTES, String.class); + verifyNoInteractions(defaultDelegate); + } + } + + @Test + void deserialize_zstdMagic_doesNotTouchDefaultDelegate() { + try (MockedStatic mocked = mockStatic(Utils.ZstdUtils.class)) { + mocked.when(() -> Utils.ZstdUtils.isZstd(ZSTD_BYTES)).thenReturn(true); + when(zstdDelegate.deserialize(any(), any())).thenReturn(new Object()); + + delegate.deserialize(ZSTD_BYTES, Object.class); + + verifyNoInteractions(defaultDelegate); + } + } + + // fallback + @Test + void deserialize_noZstdMagic_usesDefaultDelegate() { + try (MockedStatic mocked = mockStatic(Utils.ZstdUtils.class)) { + mocked.when(() -> Utils.ZstdUtils.isZstd(PLAIN_BYTES)).thenReturn(false); + when(defaultDelegate.deserialize(PLAIN_BYTES, String.class)).thenReturn("plain-result"); + + String result = delegate.deserialize(PLAIN_BYTES, String.class); + + assertEquals("plain-result", result); + verify(defaultDelegate).deserialize(PLAIN_BYTES, String.class); + verifyNoInteractions(zstdDelegate); + } + } + + @Test + void deserialize_noZstdMagic_doesNotTouchZstdDelegate() { + try (MockedStatic mocked = mockStatic(Utils.ZstdUtils.class)) { + mocked.when(() -> Utils.ZstdUtils.isZstd(PLAIN_BYTES)).thenReturn(false); + when(defaultDelegate.deserialize(any(), any())).thenReturn(new Object()); + + delegate.deserialize(PLAIN_BYTES, Object.class); + + verifyNoInteractions(zstdDelegate); + } + } + + @Test + void deserialize_nullBytes_routedToDefaultDelegate() { + try (MockedStatic mocked = mockStatic(Utils.ZstdUtils.class)) { + mocked.when(() -> Utils.ZstdUtils.isZstd(null)).thenReturn(false); + when(defaultDelegate.deserialize(null, String.class)).thenReturn("fallback"); + + String result = delegate.deserialize(null, String.class); + + assertEquals("fallback", result); + verify(defaultDelegate).deserialize(null, String.class); + verifyNoInteractions(zstdDelegate); + } + } + + // exceptions propagation + @Test + void deserialize_zstdDelegateThrows_exceptionPropagates() { + try (MockedStatic mocked = mockStatic(Utils.ZstdUtils.class)) { + mocked.when(() -> Utils.ZstdUtils.isZstd(ZSTD_BYTES)).thenReturn(true); + when(zstdDelegate.deserialize(any(), any())) + .thenThrow(new RuntimeException("decompression error")); + + RuntimeException ex = assertThrows(RuntimeException.class, + () -> delegate.deserialize(ZSTD_BYTES, String.class)); + assertEquals("decompression error", ex.getMessage()); + } + } + + @Test + void deserialize_defaultDelegateThrows_exceptionPropagates() { + try (MockedStatic mocked = mockStatic(Utils.ZstdUtils.class)) { + mocked.when(() -> Utils.ZstdUtils.isZstd(PLAIN_BYTES)).thenReturn(false); + when(defaultDelegate.deserialize(any(), any())) + .thenThrow(new RuntimeException("thrift error")); + + RuntimeException ex = assertThrows(RuntimeException.class, + () -> delegate.deserialize(PLAIN_BYTES, String.class)); + assertEquals("thrift error", ex.getMessage()); + } + } + + // delegation chain + @Test + void deserialize_isZstdDeterminesRouting_trueThenFalse() { + try (MockedStatic mocked = mockStatic(Utils.ZstdUtils.class)) { + byte[] bytes = PLAIN_BYTES; + + // First call: treated as Zstd + mocked.when(() -> Utils.ZstdUtils.isZstd(bytes)).thenReturn(true); + when(zstdDelegate.deserialize(bytes, String.class)).thenReturn("via-zstd"); + assertEquals("via-zstd", delegate.deserialize(bytes, String.class)); + + // Second call: treated as non-Zstd + mocked.when(() -> Utils.ZstdUtils.isZstd(bytes)).thenReturn(false); + when(defaultDelegate.deserialize(bytes, String.class)).thenReturn("via-default"); + assertEquals("via-default", delegate.deserialize(bytes, String.class)); + } + } +} diff --git a/storm-client/test/jvm/org/apache/storm/serialization/ZstdThriftSerializationDelegateTest.java b/storm-client/test/jvm/org/apache/storm/serialization/ZstdThriftSerializationDelegateTest.java new file mode 100644 index 00000000000..24faf86e153 --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/serialization/ZstdThriftSerializationDelegateTest.java @@ -0,0 +1,186 @@ +/* + * 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.storm.serialization; + +import static org.junit.jupiter.api.Assertions.assertArrayEquals; +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import org.apache.storm.Config; +import org.apache.storm.generated.StormTopology; +import org.apache.storm.utils.Utils; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +class ZstdThriftSerializationDelegateTest { + + private ZstdThriftSerializationDelegate delegate; + + private static StormTopology validTopology() { + StormTopology t = new StormTopology(); + t.set_spouts(Collections.emptyMap()); + t.set_bolts(Collections.emptyMap()); + t.set_state_spouts(Collections.emptyMap()); + return t; + } + + private static final StormTopology topology = validTopology(); + + @BeforeEach + void setUp() { + delegate = new ZstdThriftSerializationDelegate(); + delegate.prepare(Collections.emptyMap()); // uses defaults + } + + @Test + void prepare_emptyConf_usesDefaults() { + // Verify that prepare with empty conf doesn't throw and produces + // a delegate that can still serialize/deserialize correctly. + ZstdThriftSerializationDelegate d = new ZstdThriftSerializationDelegate(); + assertDoesNotThrow(() -> d.prepare(Collections.emptyMap())); + + byte[] serialized = d.serialize(topology); + assertTrue(Utils.ZstdUtils.isZstd(serialized)); + } + + @Test + void prepare_customLevel_isUsed() { + Map conf = new HashMap<>(); + conf.put(Config.STORM_COMPRESSION_ZSTD_LEVEL, 1); + + ZstdThriftSerializationDelegate d = new ZstdThriftSerializationDelegate(); + d.prepare(conf); + + // Both level 1 and level 3 produce valid zstd output — verify it is zstd and round-trips + byte[] serialized = d.serialize(topology); + assertTrue(Utils.ZstdUtils.isZstd(serialized)); + assertNotNull(d.deserialize(serialized, StormTopology.class)); + } + + @Test + void prepare_customMaxDecompressedBytes_isRespected() { + // Set a limit of 1 byte — decompressing anything non-trivial must fail + Map conf = new HashMap<>(); + conf.put(Config.STORM_COMPRESSION_ZSTD_MAX_DECOMPRESSED_BYTES, 1); + + ZstdThriftSerializationDelegate d = new ZstdThriftSerializationDelegate(); + d.prepare(conf); + + // Serialize with the default delegate (no limit) to get valid zstd bytes + byte[] serialized = delegate.serialize(topology); + + RuntimeException ex = assertThrows(RuntimeException.class, + () -> d.deserialize(serialized, StormTopology.class)); + assertEquals("Zstd decompression failed", ex.getMessage()); + } + + @Test + void serialize_tBaseObject_producesZstdFrame() { + byte[] result = delegate.serialize(topology); + + assertNotNull(result); + assertTrue(result.length > 0); + assertTrue(Utils.ZstdUtils.isZstd(result), "Serialized output must start with zstd magic"); + } + + @Test + void serialize_nonTBaseObject_throwsIllegalArgumentException() { + IllegalArgumentException ex = assertThrows(IllegalArgumentException.class, + () -> delegate.serialize("not a TBase")); + assertEquals("Object must be an instance of TBase", ex.getMessage()); + } + + @Test + void serialize_nullObject_throwsIllegalArgumentException() { + assertThrows(IllegalArgumentException.class, + () -> delegate.serialize(null)); + } + + @Test + void serialize_sameTBaseObject_producesDeterministicOutput() { + byte[] first = delegate.serialize(topology); + byte[] second = delegate.serialize(topology); + assertArrayEquals(first, second, "Serialization of the same object must be deterministic"); + } + + @Test + void deserialize_validZstdBytes_returnsCorrectType() { + byte[] serialized = delegate.serialize(topology); + + StormTopology result = delegate.deserialize(serialized, StormTopology.class); + + assertNotNull(result); + } + + @Test + void deserialize_corruptedBytes_throwsRuntimeException() { + byte[] garbage = {0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07}; + RuntimeException ex = assertThrows(RuntimeException.class, + () -> delegate.deserialize(garbage, StormTopology.class)); + assertEquals("Cannot deserialize [" +StormTopology.class.getSimpleName() + "]. " + + "Expected zstd compressed bytes, but received unknown format.", ex.getMessage()); + } + + @Test + void deserialize_wrongTargetClass_throwsRuntimeException() { + byte[] serialized = delegate.serialize(validTopology()); + assertThrows(RuntimeException.class, + () -> delegate.deserialize(serialized, String.class)); + } + + @Test + void roundTrip_emptyTopology_isLossless() { + StormTopology original = topology; + byte[] serialized = delegate.serialize(original); + StormTopology recovered = delegate.deserialize(serialized, StormTopology.class); + + assertEquals(original, recovered); + } + + @Test + void roundTrip_afterPrepareWithCustomConf_isLossless() { + Map conf = new HashMap<>(); + conf.put(Config.STORM_COMPRESSION_ZSTD_LEVEL, 9); + conf.put(Config.STORM_COMPRESSION_ZSTD_MAX_DECOMPRESSED_BYTES, 100); + + ZstdThriftSerializationDelegate d = new ZstdThriftSerializationDelegate(); + d.prepare(conf); + + StormTopology original = topology; + byte[] serialized = d.serialize(original); + StormTopology recovered = d.deserialize(serialized, StormTopology.class); + + assertEquals(original, recovered); + } + + @Test + void roundTrip_prepareNotCalled_throwsNullPointerOrRuntime() { + // Without prepare(), topoConf fields are zero/null — behaviour is defined by ObjectReader. + // At minimum, serialize must not silently corrupt data. + ZstdThriftSerializationDelegate unprepared = new ZstdThriftSerializationDelegate(); + // Either it works (ObjectReader tolerates null map) or throws — must not return wrong data. + try { + byte[] serialized = unprepared.serialize(topology); + StormTopology recovered = unprepared.deserialize(serialized, StormTopology.class); + assertNotNull(recovered); + } catch (RuntimeException e) { + // Acceptable: prepare() was never called + } + } +} diff --git a/storm-client/test/jvm/org/apache/storm/utils/UtilsTest.java b/storm-client/test/jvm/org/apache/storm/utils/UtilsTest.java index c07a054238d..d29f41d501f 100644 --- a/storm-client/test/jvm/org/apache/storm/utils/UtilsTest.java +++ b/storm-client/test/jvm/org/apache/storm/utils/UtilsTest.java @@ -20,6 +20,7 @@ import java.io.IOException; import java.net.SocketException; +import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -517,4 +518,204 @@ public void testHandleUncaughtExceptionSwallowsCausedAndDerivedExceptions() { fail(expectationMessage, unexpected); } } + + @Test + void compress_nullInput_returnsNull() { + assertNull(Utils.ZstdUtils.compress(null, 3)); + } + + @Test + void compress_emptyInput_returnsEmptyArray() { + byte[] result = Utils.ZstdUtils.compress(new byte[0], 3); + assertNotNull(result); + assertEquals(0, result.length); + } + + @Test + void compress_singleByte_producesZstdFrame() { + byte[] input = {0x42}; + byte[] compressed = Utils.ZstdUtils.compress(input, 3); + assertTrue(Utils.ZstdUtils.isZstd(compressed), + "Compressed output should start with the Zstd magic number"); + } + + @Test + void compress_smallPayload_roundtrips() { + byte[] input = "Hello, ZSTD!".getBytes(StandardCharsets.UTF_8); + byte[] compressed = Utils.ZstdUtils.compress(input, 3); + byte[] decompressed = Utils.ZstdUtils.decompress(compressed, 1024 * 1024); + assertArrayEquals(input, decompressed); + } + + @Test + void compress_largeRepetitivePayload_isSmallerThanOriginal() { + // Highly compressible: 64 KB of zeros + byte[] input = new byte[64 * 1024]; + byte[] compressed = Utils.ZstdUtils.compress(input, 3); + assertTrue(compressed.length < input.length, + "Compressed size should be smaller than original for repetitive data"); + } + + void compress_variousLevels_roundtrip() { + byte[] input = zstdSampleData(4096); + for (int level : Set.of(1, 3, 9, 19)) { + byte[] compressed = Utils.ZstdUtils.compress(input, level); + byte[] decompressed = Utils.ZstdUtils.decompress(compressed, input.length * 2); + assertArrayEquals(input, decompressed, + "Round-trip must be lossless at compression level " + level); + } + } + + @Test + void compress_highEntropData_doesNotThrow() { + // Random-ish bytes — Zstd may not shrink them, but must not fail + byte[] input = zstdSampleData(8192); + assertDoesNotThrow(() -> Utils.ZstdUtils.compress(input, 3)); + } + + @Test + void decompress_nullInput_returnsNull() { + assertNull(Utils.ZstdUtils.decompress(null, 1024 * 1024)); + } + + @Test + void decompress_emptyInput_returnsEmptyArray() { + byte[] result = Utils.ZstdUtils.decompress(new byte[0], 1024 * 1024); + assertNotNull(result); + assertEquals(0, result.length); + } + + @Test + void decompress_validFrame_recoversOriginal() { + byte[] original = "Unit test payload".getBytes(StandardCharsets.UTF_8); + byte[] compressed = Utils.ZstdUtils.compress(original, 3); + byte[] recovered = Utils.ZstdUtils.decompress(compressed, 1024 * 1024); + assertArrayEquals(original, recovered); + } + + @Test + void decompress_corruptedData_throwsRuntimeException() { + byte[] garbage = {0x00, 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07}; + RuntimeException ex = assertThrows(RuntimeException.class, + () -> Utils.ZstdUtils.decompress(garbage, 1024 * 1024)); + assertTrue(ex.getMessage().contains("Zstd decompression failed"), + "Exception message should describe the failure"); + } + + @Test + void decompress_exceedsMaxBytes_throwsRuntimeException() { + // Compress a 4 KB payload but only allow 10 bytes out + byte[] input = zstdSampleData(4096); + byte[] compressed = Utils.ZstdUtils.compress(input, 3); + + RuntimeException ex = assertThrows(RuntimeException.class, + () -> Utils.ZstdUtils.decompress(compressed, 10)); + assertTrue(ex.getMessage().contains("Zstd decompression failed"), + "Should throw when decompressed size exceeds the limit"); + assertTrue(ex.getCause().getMessage().contains("Decompression threshold exceeded"), + "Root cause should describe the threshold breach"); + } + + @Test + void decompress_exactlyAtLimit_succeeds() { + byte[] input = "abc".getBytes(StandardCharsets.UTF_8); // 3 bytes + byte[] compressed = Utils.ZstdUtils.compress(input, 3); + // Allow exactly 3 bytes — should succeed + byte[] result = Utils.ZstdUtils.decompress(compressed, 3); + assertArrayEquals(input, result); + } + + @Test + void decompress_truncatedFrame_throwsRuntimeException() { + byte[] input = zstdSampleData(256); + byte[] compressed = Utils.ZstdUtils.compress(input, 3); + // Slice off the last third to simulate truncation + byte[] truncated = new byte[compressed.length * 2 / 3]; + System.arraycopy(compressed, 0, truncated, 0, truncated.length); + + assertThrows(RuntimeException.class, + () -> Utils.ZstdUtils.decompress(truncated, 1024 * 1024)); + } + + @Test + void isZstd_nullInput_returnsFalse() { + assertFalse(Utils.ZstdUtils.isZstd(null)); + } + + @Test + void isZstd_emptyArray_returnsFalse() { + assertFalse(Utils.ZstdUtils.isZstd(new byte[0])); + } + + @Test + void isZstd_arrayTooShort_returnsFalse() { + // Only 3 bytes — not enough for the 4-byte magic + assertFalse(Utils.ZstdUtils.isZstd(new byte[]{(byte) 0x28, (byte) 0xB5, (byte) 0x2F})); + } + + @Test + void isZstd_validMagicBytes_returnsTrue() { + // Pad with enough trailing bytes so INT_HANDLE.get(bytes, 0) does not throw + byte[] magic = new byte[]{(byte) 0x28, (byte) 0xB5, (byte) 0x2F, (byte) 0xFD, 0x00}; + assertTrue(Utils.ZstdUtils.isZstd(magic)); + } + + @Test + void isZstd_validCompressedOutput_returnsTrue() { + byte[] compressed = Utils.ZstdUtils.compress("test".getBytes(StandardCharsets.UTF_8), 3); + assertTrue(Utils.ZstdUtils.isZstd(compressed), + "Output of compress() must be recognised as Zstd"); + } + + @Test + void isZstd_plainText_returnsFalse() { + byte[] plain = "Hello, world!".getBytes(StandardCharsets.UTF_8); + assertFalse(Utils.ZstdUtils.isZstd(plain)); + } + + @Test + void isZstd_wrongMagicOrder_returnsFalse() { + // Big-endian order of the same bytes — should NOT match + byte[] wrongEndian = {(byte) 0xFD, (byte) 0x2F, (byte) 0xB5, (byte) 0x28, 0x00}; + assertFalse(Utils.ZstdUtils.isZstd(wrongEndian)); + } + + @Test + void isZstd_almostCorrectMagic_returnsFalse() { + // One byte off + byte[] close = {(byte) 0x28, (byte) 0xB5, (byte) 0x2F, (byte) 0xFE, 0x00}; + assertFalse(Utils.ZstdUtils.isZstd(close)); + } + + @Test + void roundTrip_emptyStringBytes_succeeds() { + byte[] input = "".getBytes(StandardCharsets.UTF_8); + // compress() returns the original array for empty input + byte[] compressed = Utils.ZstdUtils.compress(input, 3); + assertArrayEquals(input, compressed); // empty → empty, no compression + } + + @Test + void roundTrip_binaryData_succeeds() { + byte[] input = zstdSampleData(1024); + byte[] compressed = Utils.ZstdUtils.compress(input, 3); + byte[] decompressed = Utils.ZstdUtils.decompress(compressed, input.length * 2); + assertArrayEquals(input, decompressed); + } + + @Test + void roundTrip_unicodePayload_succeeds() { + byte[] input = "apache storm - storm - apache".getBytes(StandardCharsets.UTF_8); + byte[] compressed = Utils.ZstdUtils.compress(input, 3); + byte[] decompressed = Utils.ZstdUtils.decompress(compressed, input.length * 4); + assertArrayEquals(input, decompressed); + } + + private static byte[] zstdSampleData(int size) { + byte[] data = new byte[size]; + for (int i = 0; i < size; i++) { + data[i] = (byte) (i % 256); + } + return data; + } } diff --git a/storm-server/src/test/java/org/apache/storm/daemon/supervisor/BasicContainerTest.java b/storm-server/src/test/java/org/apache/storm/daemon/supervisor/BasicContainerTest.java index b458c9164a9..a4f6ec93aa1 100644 --- a/storm-server/src/test/java/org/apache/storm/daemon/supervisor/BasicContainerTest.java +++ b/storm-server/src/test/java/org/apache/storm/daemon/supervisor/BasicContainerTest.java @@ -322,7 +322,7 @@ public void testLaunch() throws Exception { st.set_spouts(new HashMap<>()); st.set_bolts(new HashMap<>()); st.set_state_spouts(new HashMap<>()); - byte[] serializedState = Utils.gzip(Utils.thriftSerialize(st)); + byte[] serializedState = Utils.GzipUtils.compress(Utils.thriftSerialize(st)); final Map superConf = new HashMap<>(); superConf.put(Config.STORM_LOCAL_DIR, stormLocal); @@ -426,7 +426,7 @@ public void testLaunchStorm1version() throws Exception { // minimum 1.x version of supporting STORM-2448 would be 1.0.4 st.set_storm_version("1.0.4"); - byte[] serializedState = Utils.gzip(Utils.thriftSerialize(st)); + byte[] serializedState = Utils.GzipUtils.compress(Utils.thriftSerialize(st)); final Map superConf = new HashMap<>(); superConf.put(Config.STORM_LOCAL_DIR, stormLocal); @@ -529,7 +529,7 @@ public void testLaunchStorm0version() throws Exception { // minimum 0.x version of supporting STORM-2448 would be 0.10.3 st.set_storm_version("0.10.3"); - byte[] serializedState = Utils.gzip(Utils.thriftSerialize(st)); + byte[] serializedState = Utils.GzipUtils.compress(Utils.thriftSerialize(st)); final Map superConf = new HashMap<>(); superConf.put(Config.STORM_LOCAL_DIR, stormLocal);