This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new 8001ca972 Zstd compression for thrift serialization (storm cluster
state) (#8653)
8001ca972 is described below
commit 8001ca97284141c5ad8c6c77ce73f7249cf48f34
Author: Gianluca Graziadei <[email protected]>
AuthorDate: Wed May 20 19:39:45 2026 +0200
Zstd compression for thrift serialization (storm cluster state) (#8653)
* zstd compression for thrift serialization
* rename zstd compression level parm
* remove zstd-jni compile scope
* remove zstd-jni compile scope
* delegation chain, manage exceptions, manage zip bomb, unit test coverage
* increase to 100MB default max uncompressed topology bytes
* docs
* add `ZstdBridgeThriftSerializationDelegateRoundTripTest`
---
conf/defaults.yaml | 5 +-
docs/Cluster-State-Serialization.md | 124 +++++++++++
docs/Serialization.md | 2 +
pom.xml | 6 +
storm-client/pom.xml | 8 +
storm-client/src/jvm/org/apache/storm/Config.java | 19 +-
.../GzipBridgeThriftSerializationDelegate.java | 2 +-
.../serialization/GzipSerializationDelegate.java | 35 ++-
.../GzipThriftSerializationDelegate.java | 12 +-
.../ZstdBridgeThriftSerializationDelegate.java | 50 +++++
.../ZstdThriftSerializationDelegate.java | 87 ++++++++
.../src/jvm/org/apache/storm/utils/Utils.java | 201 +++++++++++++++--
.../apache/storm/validation/ConfigValidation.java | 20 ++
.../GzipBridgeThriftSerializationDelegateTest.java | 2 +
...eThriftSerializationDelegateRoundTripTest.java} | 50 ++---
.../ZstdBridgeThriftSerializationDelegateTest.java | 245 +++++++++++++++++++++
.../ZstdThriftSerializationDelegateTest.java | 186 ++++++++++++++++
.../test/jvm/org/apache/storm/utils/UtilsTest.java | 201 +++++++++++++++++
.../daemon/supervisor/BasicContainerTest.java | 6 +-
19 files changed, 1191 insertions(+), 70 deletions(-)
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index bc9a5979d..2c3bb9e06 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 000000000..8581448f4
--- /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 4d2749105..0e7cc2e87 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 326bf4543..d98a27c85 100644
--- a/pom.xml
+++ b/pom.xml
@@ -82,6 +82,7 @@
<!-- dependency versions -->
<commons-compress.version>1.28.0</commons-compress.version>
+ <zstd-jni.version>1.5.7-8</zstd-jni.version>
<commons-io.version>2.22.0</commons-io.version>
<commons-lang3.version>3.20.0</commons-lang3.version>
<commons-exec.version>1.6.0</commons-exec.version>
@@ -514,6 +515,11 @@
<artifactId>commons-compress</artifactId>
<version>${commons-compress.version}</version>
</dependency>
+ <dependency>
+ <groupId>com.github.luben</groupId>
+ <artifactId>zstd-jni</artifactId>
+ <version>${zstd-jni.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-exec</artifactId>
diff --git a/storm-client/pom.xml b/storm-client/pom.xml
index 62d6c20d9..3a7ee315d 100644
--- a/storm-client/pom.xml
+++ b/storm-client/pom.xml
@@ -135,6 +135,14 @@
<artifactId>curator-test</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.commons</groupId>
+ <artifactId>commons-compress</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.github.luben</groupId>
+ <artifactId>zstd-jni</artifactId>
+ </dependency>
</dependencies>
<build>
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java
b/storm-client/src/jvm/org/apache/storm/Config.java
index 62770b62e..0d7046585 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -1485,7 +1485,24 @@ public class Config extends HashMap<String, Object> {
*/
@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.
+ * <b>Prohibited:</b> 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 bc9661124..67ae15ada 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 9c4045854..d12c2b018 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.io.ObjectOutputStream;
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<String, Object> 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 class GzipSerializationDelegate implements
SerializationDelegate {
@Override
public <T> T deserialize(byte[] bytes, Class<T> 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 f518628a8..ae3bc7ac1 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 @@ import org.apache.storm.utils.Utils;
*/
public class GzipThriftSerializationDelegate implements SerializationDelegate {
+ private static final int DEFAULT_MAX_DECOMPRESSED_BYTES = 100 * 1024 *
1024;
+ private int maxDecompressedBytes;
+
@Override
public void prepare(Map<String, Object> 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 class GzipThriftSerializationDelegate implements
SerializationDelegate {
public <T> T deserialize(byte[] bytes, Class<T> 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 000000000..a3484b7fd
--- /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<String, Object> 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> T deserialize(byte[] bytes, Class<T> 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 000000000..d3b990a7b
--- /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<String, Object> 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> T deserialize(byte[] bytes, Class<T> 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 b639863f8..7ba3298fe 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.OutputStreamWriter;
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.URLDecoder;
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.GZIPOutputStream;
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.com.google.common.collect.Maps;
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 class Utils {
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.
+ *
+ * <p>The output includes the standard Zstandard frame header, making
it
+ * self-describing for the decompression phase.</p>
+ *
+ * @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 0206ec458..7f2d2462c 100644
--- a/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
+++ b/storm-client/src/jvm/org/apache/storm/validation/ConfigValidation.java
@@ -804,6 +804,26 @@ public class ConfigValidation {
}
}
+ 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 7a5634dcc..47c9147d5 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/GzipBridgeThriftSerializationDelegateTest.java
b/storm-client/test/jvm/org/apache/storm/serialization/ZstdBridgeThriftSerializationDelegateRoundTripTest.java
similarity index 56%
copy from
storm-client/test/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegateTest.java
copy to
storm-client/test/jvm/org/apache/storm/serialization/ZstdBridgeThriftSerializationDelegateRoundTripTest.java
index 7a5634dcc..eac87d844 100644
---
a/storm-client/test/jvm/org/apache/storm/serialization/GzipBridgeThriftSerializationDelegateTest.java
+++
b/storm-client/test/jvm/org/apache/storm/serialization/ZstdBridgeThriftSerializationDelegateRoundTripTest.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
@@ -12,54 +12,52 @@
package org.apache.storm.serialization;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.util.Collections;
import org.apache.storm.generated.GlobalStreamId;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-
-
-public class GzipBridgeThriftSerializationDelegateTest {
- SerializationDelegate testDelegate;
+class ZstdBridgeThriftSerializationDelegateRoundTripTest {
+ SerializationDelegate bridge;
@BeforeEach
- public void setUp() throws Exception {
- testDelegate = new GzipBridgeThriftSerializationDelegate();
+ public void setUp() {
+ bridge = new ZstdBridgeThriftSerializationDelegate();
+ bridge.prepare(Collections.emptyMap());
}
@Test
public void testDeserialize_readingFromGzip() {
GlobalStreamId id = new GlobalStreamId("first", "second");
-
byte[] serialized = new
GzipThriftSerializationDelegate().serialize(id);
- GlobalStreamId id2 = testDelegate.deserialize(serialized,
GlobalStreamId.class);
+ GlobalStreamId out = bridge.deserialize(serialized,
GlobalStreamId.class);
- assertEquals(id2.get_componentId(), id.get_componentId());
- assertEquals(id2.get_streamId(), id.get_streamId());
+ assertEquals(id.get_componentId(), out.get_componentId());
+ assertEquals(id.get_streamId(), out.get_streamId());
}
@Test
- public void testDeserialize_readingFromGzipBridge() {
- GlobalStreamId id = new GlobalStreamId("first", "second");
-
- byte[] serialized = new
GzipBridgeThriftSerializationDelegate().serialize(id);
+ public void testDeserialize_readingFromRawThrift() {
+ GlobalStreamId id = new GlobalStreamId("A", "B");
+ byte[] serialized = new ThriftSerializationDelegate().serialize(id);
- GlobalStreamId id2 = testDelegate.deserialize(serialized,
GlobalStreamId.class);
+ GlobalStreamId out = bridge.deserialize(serialized,
GlobalStreamId.class);
- assertEquals(id2.get_componentId(), id.get_componentId());
- assertEquals(id2.get_streamId(), id.get_streamId());
+ assertEquals(id.get_componentId(), out.get_componentId());
+ assertEquals(id.get_streamId(), out.get_streamId());
}
@Test
- public void testDeserialize_readingFromDefault() {
- GlobalStreamId id = new GlobalStreamId("A", "B");
-
- byte[] serialized = new ThriftSerializationDelegate().serialize(id);
+ public void testRoundTrip_throughBridge() {
+ GlobalStreamId id = new GlobalStreamId("x", "y");
+ byte[] serialized = bridge.serialize(id);
- GlobalStreamId id2 = testDelegate.deserialize(serialized,
GlobalStreamId.class);
+ GlobalStreamId out = bridge.deserialize(serialized,
GlobalStreamId.class);
- assertEquals(id2.get_componentId(), id.get_componentId());
- assertEquals(id2.get_streamId(), id.get_streamId());
+ assertEquals(id.get_componentId(), out.get_componentId());
+ assertEquals(id.get_streamId(), out.get_streamId());
}
}
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 000000000..13bf7515c
--- /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<String, Object> 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<String, Object> 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<Utils.ZstdUtils> 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<Utils.ZstdUtils> 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<Utils.ZstdUtils> 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<Utils.ZstdUtils> 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<Utils.ZstdUtils> 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<Utils.ZstdUtils> 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<Utils.ZstdUtils> 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<Utils.ZstdUtils> 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 000000000..24faf86e1
--- /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<String, Object> 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<String, Object> 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<String, Object> 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 c07a05423..d29f41d50 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 @@ package org.apache.storm.utils;
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 class UtilsTest {
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 b458c9164..a4f6ec93a 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 class BasicContainerTest {
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<String, Object> superConf = new HashMap<>();
superConf.put(Config.STORM_LOCAL_DIR, stormLocal);
@@ -426,7 +426,7 @@ public class BasicContainerTest {
// 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<String, Object> superConf = new HashMap<>();
superConf.put(Config.STORM_LOCAL_DIR, stormLocal);
@@ -529,7 +529,7 @@ public class BasicContainerTest {
// 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<String, Object> superConf = new HashMap<>();
superConf.put(Config.STORM_LOCAL_DIR, stormLocal);
