docs/Cluster-State-Serialization.md

---
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.

> **Note on worker heartbeats.** Since 2.0 ([STORM-2693](https://issues.apache.org/jira/browse/STORM-2693)),
> worker liveness heartbeats are, by default, *not* persisted in ZooKeeper: workers
> write them to local disk, supervisors relay them to Nimbus over Thrift, and Nimbus
> keeps them in an in-memory heartbeat cache. Worker heartbeats are only written to a
> state store (the `WORKERBEATS_SUBTREE` path) when a heartbeat store such as Pacemaker
> is configured. The serialization described below still applies to those stored
> heartbeats, and to supervisor liveness (`SupervisorInfo`), which is always kept as an
> ephemeral ZooKeeper node.

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.

## Heartbeat timestamps and the year 2038

Since STORM-7897, the `time_secs` field of `ClusterWorkerHeartbeat`,
`SupervisorWorkerHeartbeat` and `LSWorkerHeartbeat` is a 64-bit integer
(`i64`), and all heartbeat writers and timeout checks use the long-based
clock (`Time.currentTimeSecsLong()` / `Time.deltaSecsLong(...)`). Earlier
releases carried these timestamps as `i32` seconds, which overflows on
2038-01-19T03:14:07Z and would have caused Nimbus to treat live workers
as dead.

`uptime_secs` fields remain `i32`: they are relative durations, not
absolute timestamps. `Time.currentTimeSecs()` and `Time.deltaSecs(int)`
are deprecated but retained for such relative-duration callers.

### Upgrade implications

Thrift tags `i32` and `i64` values differently on the wire, so heartbeat
blobs written by a pre-upgrade daemon do **not** deserialize under the
new schema: the reader skips the mistyped field and the blob then fails
required-field validation with a `TProtocolException`. In practice:

1. **A full-cluster upgrade is required.** Do not run a mixed-version
   cluster across Nimbus, Supervisors and workers: heartbeats do not
   round-trip between the old and new schema in either direction.
2. **In-flight heartbeats are dropped once, then self-heal.** Nimbus
   times workers out by *receipt* time, so a dropped heartbeat is
   replaced on the next report interval; supervisors rewrite their
   on-disk `LSWorkerHeartbeat` local state the same way. Expect at most
   one report cycle of staleness around the restart, which is within the
   normal tolerance of a full-cluster bounce.