This is an automated email from the ASF dual-hosted git repository.
reiabreu 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 0cf51b679 Add tuple compression for inter-worker communication (#8707)
0cf51b679 is described below
commit 0cf51b6793b4f7eee02f09034992f3215ee7804b
Author: Gianluca Graziadei <[email protected]>
AuthorDate: Sat May 30 17:31:59 2026 +0200
Add tuple compression for inter-worker communication (#8707)
* init config
* (de)serialization logic
* deserialization logic refinement, unit tests
* docs + storm perf example
* apply new checkstyle rules
* config param `topology.tuple.compression.max.decompressed.bytes`
* improve deserialize zstd false positive collision logic
* enter the decompress branch when the topology actually uses compression
* minor changes
* docs changes + additional test case
* remove unnecessary arrays allocation in `KryoTupleSerializer`
* add LOG.debug for deserialize collision
* ad-hoc test cases for false positive isZstd deserializer
* add bench in docs
---
conf/defaults.yaml | 3 +
docs/Serialization.md | 83 +++++
.../FileReadWordCountSpoutCompressionTopo.java | 94 ++++++
.../src/main/sampledata/longrandomwords.txt | 50 +++
storm-client/src/jvm/org/apache/storm/Config.java | 26 ++
.../storm/serialization/KryoTupleDeserializer.java | 64 +++-
.../storm/serialization/KryoTupleSerializer.java | 29 +-
.../src/jvm/org/apache/storm/utils/Utils.java | 40 ++-
.../KryoTupleSerializerDeserializerTest.java | 373 +++++++++++++++++++++
.../test/jvm/org/apache/storm/utils/UtilsTest.java | 50 +++
10 files changed, 797 insertions(+), 15 deletions(-)
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index 2c3bb9e06..d5d6bb451 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -55,9 +55,12 @@ 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.ZstdBridgeThriftSerializationDelegate"
+topology.tuple.compression.threshold: 1460
+topology.tuple.compression.enable: false
storm.compression.zstd.level: 3
storm.compression.zstd.max.decompressed.bytes: 104857600
storm.compression.gzip.max.decompressed.bytes: 104857600
+topology.tuple.compression.max.decompressed.bytes: 10485760
storm.codedistributor.class:
"org.apache.storm.codedistributor.LocalFileSystemCodeDistributor"
storm.workers.artifacts.dir: "workers-artifacts"
storm.health.check.dir: "healthchecks"
diff --git a/docs/Serialization.md b/docs/Serialization.md
index 0e7cc2e87..e7af57477 100644
--- a/docs/Serialization.md
+++ b/docs/Serialization.md
@@ -61,6 +61,89 @@ Beware that Java serialization is extremely expensive, both
in terms of CPU cost
You can turn on/off the behavior to fall back on Java serialization by setting
the `Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION` config to true/false. The
default value is false for security reasons.
+### Tuple compression
+
+For inter-worker (remote) traffic, Storm can optionally compress serialized
tuples with [Zstandard](https://facebook.github.io/zstd/) before they are sent
over the network. This is intended for one specific scenario: components that
emit **large** payloads to a remote worker, where the bytes saved on the wire
outweigh the CPU cost of compression. A good example is a spout that emits
entire lines of text to a downstream bolt running on a different worker.
+
+Compression is **disabled by default** and follows the serialization lifecycle
exactly:
+
+- **Intra-worker (local) traffic** bypasses `KryoTupleSerializer` altogether,
so it is never compressed regardless of configuration. You do not pay any CPU
cost for tuples that stay inside a worker process.
+- **Inter-worker (remote) traffic** is compressed only when compression is
enabled for the source component *and* the serialized tuple is larger than the
configured threshold. Small tuples (single words, IDs, etc.) are left
uncompressed, since the framing overhead of a compressed payload can exceed the
original size.
+
+#### Enabling compression per component
+
+Compression is controlled by the component-specific configuration
`topology.tuple.compression.enable`. Because Storm merges component-specific
configuration over the topology configuration, you can enable it for just the
components that emit large tuples, leaving the rest of the topology untouched:
+
+```java
+TopologyBuilder builder = new TopologyBuilder();
+
+builder.setSpout(SPOUT_ID, new FileReadSpout(inputFile), spoutNum)
+ .addConfiguration(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE, true);
+
+builder.setBolt(SPLIT_ID, new SplitSentenceBolt(), spBoltNum)
+ .localOrShuffleGrouping(SPOUT_ID);
+builder.setBolt(COUNT_ID, new CountBolt(), cntBoltNum)
+ .fieldsGrouping(SPLIT_ID, new Fields(SplitSentenceBolt.FIELDS));
+```
+
+You can also enable it topology-wide (or cluster-wide via `storm.yaml`) by
setting `topology.tuple.compression.enable: true`, but enabling it only where
large tuples are actually emitted is recommended.
+
+#### Flux
+
+> **Note:** With [Flux](flux.html), only **topology-wide** enablement is
currently possible. Flux has no per-component configuration mechanism —
`FluxBuilder` applies only parallelism, number of tasks, memory/CPU load, and
groupings to the underlying declarers, and the `config:` block is
topology-scoped. There is no Flux equivalent of
`declarer.addConfiguration(...)`, so the per-component approach recommended
above cannot be expressed in a Flux YAML definition.
+
+To enable compression for a Flux topology, set it in the topology-level
`config:` block:
+
+```yaml
+config:
+ topology.tuple.compression.enable: true
+ topology.tuple.compression.threshold: 1460
+```
+
+Be aware that this enables compression for *every* remote-bound tuple in the
topology that exceeds the threshold.
+
+#### Configuration reference
+
+| Config | Default | Description |
+| --- | --- | --- |
+| `topology.tuple.compression.enable` | `false` | Enables Zstd compression of
serialized tuples before remote transfer. Best set per component via
`addConfiguration`. |
+| `topology.tuple.compression.threshold` | `1460` | Minimum serialized tuple
size, in bytes, before compression is attempted. Tuples at or below this size
are sent uncompressed. The default matches the typical Ethernet TCP MSS, so
payloads that already fit in a single network frame are never compressed. |
+| `storm.compression.zstd.level` | `3` | Zstd compression level. Supported
range is 1–19; levels 20–22 (ultra mode) are prohibited because of their memory
requirements. |
+| `topology.tuple.compression.max.decompressed.bytes` | `10485760` (10 MB) |
Upper bound on the decompressed size of a single tuple. Decompression that
would exceed this limit fails, guarding against malicious or corrupt payloads. |
+
+#### How decompression works
+
+Compression is self-describing on the wire, so **no extra configuration is
required on the receiving side**. The deserializer inspects the leading bytes
of each incoming payload: if they match the Zstd magic header it decompresses
the payload (bounded by `topology.tuple.compression.max.decompressed.bytes`)
before deserializing, otherwise it deserializes the bytes directly. A single
deserializer therefore transparently handles a mix of compressed and
uncompressed tuples.
+
+As an optimization, the deserializer determines once — when the worker starts
— whether *any* component in the topology enables compression (by scanning the
merged per-component configurations). If none does, the magic-header check is
skipped entirely and the Zstd code path is never touched, so topologies that do
not use the feature pay no per-tuple cost. The corollary is that compression
must be enabled somewhere in the topology config for compressed tuples to be
decompressed on receipt [...]
+
+#### Indicative benchmark
+
+> **Disclaimer:** These numbers were gathered in a limited capacity while
developing this feature and should be treated as a rough guide only, not as a
performance guarantee. They were produced on a specific, deliberately
favourable setup and your results will vary with topology shape, tuple size,
network characteristics, and hardware.
+
+The benchmark ran two equivalent word-count topologies defined in `storm-perf`
— one with tuple compression enabled in Spout component
(`FileReadWordCountSpoutCompressionTopo`) and one without
(`FileReadWordCountTopo`) — across workers connected by a simulated network
with **10 ms latency** and **0.5 ms jitter**. This does not represent a typical
intra-datacenter network; it deliberately emphasizes the maximum advantage the
feature can offer when configured well. The tuple size used is t [...]
+
+Sample round-trip ping between two supervisors on the Docker network:
+```
+--- cluster-supervisor2-1 ping statistics ---
+5 packets transmitted, 5 received, 0% packet loss, time 4004ms
+rtt min/avg/max/mdev = 18.767/24.353/42.486/9.083 ms
+```
+
+Results (compression vs. no compression):
+
+| Metric | Compression | No compression | Difference | Better |
+| --- | --- | --- | --- | --- |
+| Avg transfer rate (msg/s) | 776,389 | 744,544 | +31,845 (+4.3%) |
Compression |
+| Peak transfer rate (msg/s) | 805,700 | 790,300 | +15,400 | Compression |
+| Avg spout throughput (acks/s) | 98,167 | 92,844 | +5,323 (+5.8%) |
Compression |
+| Peak spout throughput (acks/s) | 100,300 | 98,666 | +1,634 | Compression |
+| Avg complete latency (ms) | 362.48 | 376.73 | -14.25 (-3.8%) | Compression |
+| Max complete latency (ms) | 366.44 | 385.72 | -19.28 | Compression |
+| Runtime stability | More consistent | More fluctuation | — | Compression |
+
+In this configuration, compression improved transfer rate and spout throughput
by roughly 4–6% and reduced complete latency by a few percent, while also
producing more consistent per-task behaviour (less jitter across tasks). The
takeaway is qualitative: when large tuples cross a high-latency link, trading
CPU for fewer bytes on the wire can pay off — but you should measure with your
own workload before enabling it broadly.
+
### Component-specific serialization registrations
Storm 0.7.0 lets you set component-specific configurations (read more about
this at [Configuration](Configuration.html)). Of course, if one component
defines a serialization that serialization will need to be available to other
bolts -- otherwise they won't be able to receive messages from that component!
diff --git
a/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountSpoutCompressionTopo.java
b/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountSpoutCompressionTopo.java
new file mode 100644
index 000000000..51c43057e
--- /dev/null
+++
b/examples/storm-perf/src/main/java/org/apache/storm/perf/FileReadWordCountSpoutCompressionTopo.java
@@ -0,0 +1,94 @@
+/*
+ * 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.perf;
+
+import java.util.Map;
+import org.apache.storm.Config;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.perf.bolt.CountBolt;
+import org.apache.storm.perf.bolt.SplitSentenceBolt;
+import org.apache.storm.perf.spout.FileReadSpout;
+import org.apache.storm.perf.utils.Helper;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.utils.Utils;
+
+/**
+ * This topo helps measure speed of word count.
+ *
+ * <p>Spout loads a file into memory on initialization, then emits the lines
in an endless loop.
+ */
+public class FileReadWordCountSpoutCompressionTopo {
+ public static final String SPOUT_ID = "spout";
+ public static final String COUNT_ID = "counter";
+ public static final String SPLIT_ID = "splitter";
+ public static final String TOPOLOGY_NAME =
"FileReadWordCountSpoutCompressionTopo";
+
+ // Config settings
+ public static final String SPOUT_NUM = "spout.count";
+ public static final String SPLIT_NUM = "splitter.count";
+ public static final String COUNT_NUM = "counter.count";
+ public static final String INPUT_FILE = "input.file";
+
+ public static final int DEFAULT_SPOUT_NUM = 1;
+ public static final int DEFAULT_SPLIT_BOLT_NUM = 2;
+ public static final int DEFAULT_COUNT_BOLT_NUM = 2;
+
+
+ static StormTopology getTopology(Map<String, Object> config) {
+
+ final int spoutNum = Helper.getInt(config, SPOUT_NUM,
DEFAULT_SPOUT_NUM);
+ final int spBoltNum = Helper.getInt(config, SPLIT_NUM,
DEFAULT_SPLIT_BOLT_NUM);
+ final int cntBoltNum = Helper.getInt(config, COUNT_NUM,
DEFAULT_COUNT_BOLT_NUM);
+ final String inputFile = Helper.getStr(config, INPUT_FILE);
+
+ TopologyBuilder builder = new TopologyBuilder();
+ // sampledata/longrandomwords.txt contains sentences with at least
1500 bytes
+ builder.setSpout(SPOUT_ID, new FileReadSpout(inputFile), spoutNum)
+ .addConfiguration(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE,
true);
+ builder.setBolt(SPLIT_ID, new SplitSentenceBolt(),
spBoltNum).localOrShuffleGrouping(SPOUT_ID);
+ builder.setBolt(COUNT_ID, new CountBolt(),
cntBoltNum).fieldsGrouping(SPLIT_ID, new Fields(SplitSentenceBolt.FIELDS));
+
+ return builder.createTopology();
+ }
+
+ public static void main(String[] args) throws Exception {
+ int runTime = -1;
+ Config topoConf = new Config();
+ if (args.length > 0) {
+ runTime = Integer.parseInt(args[0]);
+ }
+ if (args.length > 1) {
+ topoConf.putAll(Utils.findAndReadConfigFile(args[1]));
+ }
+ topoConf.put(Config.TOPOLOGY_PRODUCER_BATCH_SIZE, 1000);
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_STRATEGY,
"org.apache.storm.policy.WaitStrategyPark");
+ topoConf.put(Config.TOPOLOGY_BOLT_WAIT_PARK_MICROSEC, 0);
+ topoConf.put(Config.TOPOLOGY_DISABLE_LOADAWARE_MESSAGING, true);
+ topoConf.put(Config.TOPOLOGY_STATS_SAMPLE_RATE, 0.0005);
+
+ topoConf.putAll(Utils.readCommandLineOpts());
+ if (args.length > 2) {
+ System.err.println("args: [runDurationSec] [optionalConfFile]");
+ return;
+ }
+ // Submit topology to storm cluster
+ Helper.runOnClusterAndPrintMetrics(runTime, TOPOLOGY_NAME, topoConf,
getTopology(topoConf));
+ }
+}
diff --git a/examples/storm-perf/src/main/sampledata/longrandomwords.txt
b/examples/storm-perf/src/main/sampledata/longrandomwords.txt
new file mode 100644
index 000000000..e50c25207
--- /dev/null
+++ b/examples/storm-perf/src/main/sampledata/longrandomwords.txt
@@ -0,0 +1,50 @@
+bacterioblast neurotrophic ventricous involatile trillion oflete splenauxe
Vichyite scabbiness antiscolic unpredict mastication glacierist theologal
tetchy sapphiric unchatteled elemicin dastardliness apocalypst untongued
prefatorial Edo preparative experientialist Spatangoidea unfulminated orgiastic
pneumatotherapy Mycogone nonpoisonous bespin Jerusalem benthonic antihero
zenick supermarket divinator coadvice abscission zanyism inductivity subdentate
pneumonalgia pyrocatechol nebular fr [...]
+meloplasty Dodecatheon nonrepetition coadvice taurocolla eternal Fouquieria
retinize quadrennial provedore serpentinic planosubulate downthrust widdle
epauliere chooser ethmopalatal groundneedle ladhood elastivity cloy guitarist
bozal undiffusive eternal tristich familist counteralliance antivenin
redecrease meriquinoidal Bermudian paranephros projecting scrat scrubbed
monander propodiale planosubulate halloo emir oratorship slait eurythermal
unsupercilious tetragynian antideflation unsu [...]
+nonutilitarian galbulus snare slait pope Scanic overinstruct tramplike
Bushongo Bassaris foursquare Orbitolina rebilling corbel transcorporeal
allotropic Coniferae benthonic chordacentrum templar porriginous refasten testa
misthread euphonym naught theologal thorite blurredness starosta unstressedly
semiangle subfoliar doubtingness manny sonable seraphism archistome
prescriptible pyxie epauliere pomiferous unforkedness Kenipsim alveolite
dipsomaniacal admissory allotropic ethnocracy trop [...]
+havoc regardful angiopathy Cercosporella benzoperoxide imaginary waird
dialoguer exploiter tambo lineamental cheesecutter Scorpaenidae subdrainage
cartful embryotic unleavened uncontradictableness vinegarish quarried dosseret
beatable signifier impugnation Scanic playfellowship scotale Hydrangea
quadrennial unpatched introducer unswanlike mericarp peptonate euphemious lammy
phoenicochroite Lincolnlike figured times allotropic frenal arduousness
imprescribable Vichyite allotropic valvulot [...]
+euphonym monander psychofugal semiangle autobiographist kenno euphonym
danseuse nonutilitarian poleax biopsic placatory hackneyed tristich papery
orthopedical diopside Tsonecan chargeably biodynamics transcorporeal
Socraticism parmelioid pentosuria nonutilitarian shellworker collegian ordinant
testa antiadiaphorist twinling predisputant bot spermaphyte oratorship
cuproiodargyrite bismuthiferous ultrasystematic gallybeggar marten hackneyed
unrealize twinling velaric discipular hoove sawdu [...]
+ovopyriform rivethead Pithecolobium corbel pleasurehood dithery guanajuatite
tramplike asparaginic epididymitis arrowworm hepatorrhaphy Eryon neurotrophic
lophotrichic unstressedly Joachimite plugger trip vinegarish nonmanufacture
nigh trunnel cobeliever packsack diplomatize uninductive absvolt undangered
Itea eucalypteol tartrous weism infravaginal transude aquiline eternal
unbashfulness eurythermal theologicopolitical sviatonosite uncarefully
infestation Dadaism cuproiodargyrite biopsi [...]
+unurban docimastical Triconodonta Ghent collegian tautness figured sombreroed
pompiloid papery cockstone biodynamics ununiformly unfeeble eternal rechar
papery subfoliar plugger schoolmasterism focaloid monstership dunkadoo
parabolicness obolus heavenful antiabolitionist arrowworm dithery hellbender
myesthesia untongued chrysochrous depravity stereotypography photoelasticity
scotale overinstruct stronghearted tickleproof periarthritis experientialist
bugre preaffiliate participatingly ho [...]
+omniscribent acidophile erythrodextrin Dictograph instructiveness
experientialist focaloid basto psychofugal epidymides Homoiousian wandoo
meloplasty monilioid shellworker slipped antalgol scotale interfraternal
rotular misthread posterishness appetible misexposition arrendation
pleurotropous cocksuredom bromic Florissant ethnocracy angiolymphoma sangaree
pyroacetic Tsonecan osteopaedion unurban gunshop plerome byroad placatory
obolus noncrystallized oxyterpene lammy quintette rehabilita [...]
+myesthesia ell glandularly bugre seraphism undercolored mustafina
transcorporeal subfoliar reciprocation serphoid omniscribent redecrease
detractive terrestrially starer temporomastoid comparability Pishquow sapphiric
topline serpentinic Edo octogynous planosubulate timbermonger cacuminal weism
umangite warlike Russifier wandoo cumbrousness Whilkut silverhead predebit
bicorporeal Edo putative technopsychology Dictograph approbation noreast
uninductive ferrogoslarite prezygapophysial shal [...]
+stronghearted depressingly lithograph aprosopia parodist bought velaric
analgize epauliere ipomoein swearingly macropterous taver osteopaedion greave
hysterolysis visceral uninterpleaded piquantness benthonic fossilism
phoenicochroite devilwise bonze pamphlet preoral depravity scotale
chronographic overbuilt bot parodist poleax tristich unimmortal bought
repealableness Semecarpus louse Endomycetaceae apopenptic reformatory
tendomucoid soorkee parabolicness mesymnion orgiastic cubby recha [...]
+rivethead soorkee uncarefully collegian uncompromisingly cacuminal
cumbrousness uncompromisingness Aplacentalia subfebrile coadvice tetchy
pleurotropous boser stroking vinegarish ovoviviparous overstudiousness suspend
dishpan nigh oinomancy shibuichi coracomandibular pendulant Pithecolobium
Dawsonia gorilloid agglomeratic redesertion frictionlessly redesertion
opacousness entame pleasurehood putative epidymides morphiomania flatman
ultratense liberatress percent strander tendomucoid ting [...]
+unrevolting unobservantness stapedius tum serosanguineous clanned ventricous
times winterproof overinstruct waird reeveland metoxazine approbation aquiline
antalgol besagne mesophyte depressingly enterostomy erythrodextrin heavenful
excerpt yawler interruptor bacillite collegian sonable frenal squdge entame
galbulus Florissant propodiale cinque incomprehensible totaquina pachydermatoid
plugger mustafina unprovided umbellic preagitate unanatomized counteractively
relaster Italical spot Tr [...]
+almud Fameuse preagitate noncrystallized sterilely pseudohalogen dialoguer
alen chasmy obispo embryotic pony oxyterpene sequentially outhue propodiale
horsefly Dadaism insatiately sturdied eurythermal meloplasty Megaluridae
planosubulate bladderwort lampyrine naprapath benzoperoxide ethnocracy
porriginous stapedius ascitic unchatteled elemicin plugger pentosuria
returnability transcorporeal allectory Effie sterilely scotching underskin
equiconvex excerpt moodishness hypoid ramosopalmate [...]
+acocotl pumpkinification benzothiofuran toxihaemia boser rechar eternal
Tsonecan electrotechnics serphoid phallaceous tomorrowness winterproof
Megaluridae depthwise sloped gymnastic depressingly consumptional molossic
Consolamentum rainproof suspend phytoma glaumrie ventricous rainproof precostal
chordacentrum chacona haply eristically neurotrophic Mormyrus paleornithology
vitally lithotresis depressingly parabolicness propodiale relaster
theologicopolitical bot cubby michigan gorilloid [...]
+emir overcrown swoony molecule admissory chrysochrous impressor heavenful
ramosopalmate divinator Arneb phytoma adz tingly hymnic macropterous glacierist
tailoress euphonym scrubbed unprovided molecule astronomize bestill
hysterolysis guitarist subangulated thermochemically Cercosporella impressor
comprovincial pleasurehood stachyuraceous chilblain jirble metapolitics
quintette eulogization pachydermous barkometer nonpoisonous starer gunshop
molecule oratorize coldfinch preaffiliate Anim [...]
+daytime erythremia lophotrichic columniform thorite piquantness waird yawler
diurnalness carposporangial massedly sapience Protestantize toxihaemia
Saponaria Fameuse Pishquow wandoo hypochondriacism autoschediastical
pseudohalogen karyological Animalivora tendomucoid Semecarpus provedore
Florissant infrastapedial undinted stronghearted shola antihero folious
pneumonalgia endotheliomyoma subofficer Thraupidae Socraticism stroking
embryotic Ochnaceae erythrodextrin eurythermal pumpkinifica [...]
+anta sapience choralcelo Edo infestation mustafina cubit psychofugal affaite
inertly migrainoid craglike ovoviviparous sud hypochondriacism comparability
halloo trillium tonsure thiodiazole unleavened elastivity cacuminal
bismuthiferous suspend rivethead afterpressure feasibleness farrantly refasten
swangy physiologian transude figured Cercosporella outwealth naprapath massedly
counteractively balanocele groundneedle whitlowwort seditious topline overbuilt
rivethead Dawsonia plerome seei [...]
+uloid besagne dehairer technopsychology Sebastian orthopedical infestation
afterpressure imaginary potentness Cimmerianism pumpkinification defensibly
hogmace brag snare Vichyite alveolite triakistetrahedral unsupercilious
arduousness abscission stentorophonic steprelationship quintette
archididascalian Sphenodontidae insatiately ticktick flippantness pomiferous
ungouged bubble louse speckedness upswell hypochondriacism scotching coadvice
posterishness saccharogenic genii theologicopolit [...]
+stroking homotransplant posttraumatic silicize magnificently yawler Coniferae
wandoo gelatinousness wherefrom hogmace drome emir supraoesophageal arval
disilane Oryzorictinae penult quad unchatteled intuition edificator
predisputant cyanoguanidine predisputant playfellowship nonprofession
perfunctory infravaginal yeelaman magnetooptics Semecarpus aprosopia Glecoma
velaric knob refective friarhood friarhood prolificy merciful cyanophilous
serpentinic molossic eternal trailmaking projectin [...]
+redecrease antideflation splenauxe sertularian aurothiosulphuric supermarket
bugre Ghent shola massedly posterishness sviatonosite Hydrangea embryotic
omniscribent Confervales pondside mesymnion pleurotropous coracomandibular
isopelletierin oxyterpene archesporial fetlocked pterostigma abscission Dadaism
foursquare trisilicic inexistency relaster prefatorial tonsure Socraticism
doubtingness roughcast untongued macropterous corona whittle Sebastian jharal
deepmost unprovided unburnt byroa [...]
+bestill semantician unrevolting tartrous Lemuridae erlking naprapath testa
dishpan potentness avengeful abstractionism unprovided prefatorial
frontoorbital wandoo discipular terrestrially epididymitis knob byroad
Fouquieria flippantness spermaphyte oblongly Ochnaceae slangy silverhead
unaccessible putative downthrust Homoiousian thermochemically japanned
Ophiosaurus scrubbed leucophoenicite dermorhynchous bonze basto Dictograph
unfeeble ascitic undecorated ultraobscure figureheadship cya [...]
+Pishquow warlike biodynamics antineuritic chronographic timbermonger
floatability nonrepetition Joachimite chilblain nonpoisonous Munychian tautness
biodynamics bought antiadiaphorist valvula depravity bettermost preagitate
Prosobranchiata paranephros tetragynian antivenin rebilling cromlech
sertularian tonsure various temporomastoid oversand unfulminated ovoviviparous
autoschediastical lithotresis prolificy thermochemically aprosopia
Sphenodontidae merciful serosanguineous acidophile He [...]
+parastas Homoiousian Kenipsim phlogisticate lophotrichic naught penult
minniebush phallaceous Triphora afterpressure eternal carposporangial antivenin
rizzomed reciprocation leucophoenicite rosaniline umangite allectory marshiness
Italical vesperal analgic endotheliomyoma Spatangoidea sonable kerykeion
phytonic Munychian pondside supraoesophageal skyshine Llandovery pope pelf
componental calycular macropterous rehabilitative yote mammonish
autobiographist codisjunct Ludgatian sirrah mutt [...]
+exprobratory infestation analgic thermanesthesia deindividualization
generalizable zoonitic mastication phytoma scotching Babylonism Pincian
japanned transcorporeal imprescribable rebilling opacousness pneumatotherapy
nonexecutive Pithecolobium pamphlet thermoresistant unlapsing fossilism ungrave
bespin shellworker upswell jirble trailmaking unchatteled amender Protestantize
stiffish precostal stroking boser sleigher marten subfebrile depravity
involatile uniarticular tricae schoolmaster [...]
+massedly becomma pentafid orchiocatabasis propheticism constitutor ungouged
iniquitously sirrah prescriptible socioromantic omniscribent quailberry
generalizable bacterioblast asparaginic tartrous stradametrical redesertion
Zuludom misthread Passiflorales inductivity pendulant Pishquow unlapsing
balladmonger antalgol diminutively triradiated dipsomaniacal electrotechnics
brutism Babylonism tetchy Yannigan strammel saguran scabbiness unexplicit
regardful boor plugger Ghent biopsic trampli [...]
+Mormyrus preagitate aspersor Semecarpus lienteria subofficer mediateness Tamil
bunghole corelysis admissory bespin cyanophilous Semecarpus retinize sturdied
golem benzoperoxide consumptional disilane refasten reappreciate Mycogone
prezygapophysial scotale licitness oratorize groundneedle idiotize
sesquiquintile uniarticular serosanguineous abstractionism triakistetrahedral
flushgate louse yawler prepavement meriquinoidal amylogenesis abthainry Edo
untongued Cephalodiscus sombreroed quint [...]
+impugnation groundneedle goladar astucious sesquiquintile wemless spermaphyte
unpremonished metastoma dinical packsack ungrave yote percent dipsomaniacal
scabbiness dehairer paleornithology Lincolnlike corona bettermost Ludgatian
diopside Alethea Confervales pneumonalgia refective Alethea ungouged
commandingness unleavened Dawsonia transude glyphography pondside dosseret
subofficer rede gymnastic opacousness cretaceous unchatteled trabecular
outguess myesthesia sequentially doubtingness [...]
+zoonitic gymnastic epauliere ethnocracy scrat culm havoc unschematized phytoma
ununiformly nonprofession phytoma unforkedness manilla visceral calycular
atlantite lithotresis Whilkut frontoorbital bucketer lampyrine seizing uvanite
nativeness vesperal parmelioid pictorially stiffish calycular Scanic
subirrigate velaric subofficer diwata bespin cromlech sheepskin ramosopalmate
expiscate ambitus papery photoelasticity oblongly pendulant ploration
trailmaking blightbird foursquare arrendati [...]
+Jerusalem Inger Munychian opacousness genii licitness uncontradictableness
antalgol constitutor monogoneutic sequacity cretaceous antalgol licitness
dispermy seelful equiconvex codisjunct commotion subirrigate genii sequacity
trophonema pyxie bucketer ordinant octogynous uninterpleaded widdle lineamental
outhue monander naught limpet sviatonosite farrantly prescriptible antalgol
culm rebilling apocalypst paleornithology putative pelvimetry poleax
osteopaedion cockal elastivity trophonema [...]
+seminonflammable Bertat macropterous hypochondriacism mastication Whilkut
prospectiveness Socraticism raphis percent palaeotheriodont rizzomed trophonema
crystallographical upswell metapolitics Pishquow prescriptible squdge
overstudiousness rebilling abscission putative terrestrially scrubbed limpet
Chiasmodontidae Joachimite laubanite metoxazine phallaceous ovoviviparous
seelful spot orthopedical ploration serosanguineous golem knob gallybeggar
arval bespin laurinoxylon triakistetrahedr [...]
+antiabolitionist planispheric astucious transcorporeal timbermonger boser
volcano michigan tickleproof stachyuraceous dosseret guitarist concretion
timbermonger posterishness unpeople manganosiderite pseudohalogen homeotypical
angiopathy shibuichi overbuilt bugre parabolicness frameable hysterogen
decidable undecorated frameable botchedly infravaginal danseuse antiadiaphorist
gala lampyrine unexplicit choralcelo volcano diwata vitally involatile
sesquiquintile repealableness lampyrine la [...]
+inertly unstressedly unfurbelowed metapolitics periclitation beneficent
migrainoid semantician brag Florissant helminthagogic doubtingness overcrown
ovoviviparous gelatinousness hysterogen reconciliable visceral unswanlike bot
reappreciate technopsychology monander transcorporeal ploration scrubbed
Pishquow astucious corelysis tendomucoid socioromantic concretion topsail
bucketer preagitate phallaceous gorilloid naprapath Dictograph Megaluridae
karyological trunnel rotular Helvidian limp [...]
+undinted pondside guanajuatite Whilkut scotching pseudoxanthine choralcelo
sesquiquintile haply neurodegenerative sarcologist signifier columniform
superindifference supraoesophageal porriginous unburnt unchatteled mendacity
ovopyriform unstressedly lammy untongued pyroacetic erlking rechar angina
overinstruct pumpkinification astronomize provedore stormy hellbender analgize
mustafina redesertion pneumonalgia deaf quarried dastardliness Macraucheniidae
oversand comparability crystallogra [...]
+golem chrysochrous Macraucheniidae tetragynian timbermonger apopenptic ascitic
bestill ell nonlustrous posttraumatic sedentariness quailberry endotheliomyoma
prescriptible discipular adatom idiotize wingable excerpt Prosobranchiata
glandularly hysterolysis volcano Animalivora taver quintette scabbardless
abusiveness foursquare commotion commandingness Orbitolina Ghent cresylite
Savitar unreprimanded hellbender transude swoony trillion overcontribute
spookdom counteractively tetrahedral u [...]
+rotular monstership Scanic topsail stradametrical prezygapophysial parastas
ethmopalatal Harpa Dawsonia Hu frameable pelf ornithodelphous packsack merciful
pompiloid unstipulated furacious apopenptic saponaceous dispermy Cephalodiscus
stachyuraceous pansophism hemimelus corelysis pachydermatoid roughcast coadvice
pictorially overcontribute unharmed frontoorbital saccharogenic socioromantic
guitarist lifter pterostigma unpatched Vaishnavism reciprocation
subsequentially whittle dinical ne [...]
+subirrigate coldfinch bettermost laryngic Fameuse afterpressure mericarp sud
toxihaemia unpredict Russifier bladderwort unfulminated vinegarish hymnic
Hydrangea Coniferae Fameuse lineamental packsack weism Cercosporella Hydrangea
plerome predisputant subirrigate Quakerishly heavenful ramosopalmate by
unlapsing pompiloid prezygapophysial unpeople ventricous farrantly
planosubulate corelysis figureheadship diatomaceous octogynous dipsomaniacal
spiciferous angina evictor deaf docimastical o [...]
+cheesecutter exprobratory bestill hypoplastral japanned greave metaphonical
totaquina oratorize mammonish incomprehensible incomprehensible parmelioid
Alethea iniquitously kerykeion cloy sertularian plerome scyphostoma commotion
unchatteled chorograph Confervales monogoneutic upswell pope diminutively
ramosopalmate mastication topsail planispheric unfurbelowed ventricous
unpredict chasmy cumbrousness pony pyxie parabolicness archistome bestill
epauliere tetrahedral Muscicapa exprobratory [...]
+Bassaris raphis incomprehensible visceral overstaid erlking amplexifoliate
silverhead underskin periarthritis Dictograph chacona zenick toxihaemia
preagitate folious propodiale familist plerome psychofugal apocalypst
unrepealably seelful codisjunct zoonitic pentosuria nonuple bicorporeal haply
hackneyed halloo diminutively antalgol sombreroed eulogization unbashfulness
collegian almud sleigher photoelasticity chilblain familist foursquare
mustafina cacuminal misthread serpentinic putativ [...]
+lampyrine oversand magnetooptics Hu agglomeratic Cimmerianism redecrease
friarhood Pincian parastas gunshop farrantly involatile impairment unimmortal
uncompromisingly trillion endotheliomyoma subfebrile dehairer pondside topsail
subfebrile Pincian widdle percent seminonflammable Mormyrus noreast sturdied
massedly Babylonism orgiastic ineunt nativeness dinical Arneb charioteer
pompiloid overbuilt unchatteled blurredness debromination superindifference
symbiogenetically greave Semecarpus [...]
+swearingly monilioid phytonic bicorporeal ununiformly quintette magnetooptics
cheesecutter bogydom acidophile subdrainage subdrainage nonlustrous toplike
leucophoenicite timbermonger rebilling manny dermorhynchous silverhead
rivethead seraphism lyrebird pentafid farrantly analgize valvula elemicin pyxie
glyphography jirble overstudiousness sural aurothiosulphuric tricae
magnetooptics Hydrangea botchedly pictorially ovopyriform bladderwort
phallaceous divinator consumptional nebular terre [...]
+theologal uloid Hu metapolitics Bertat unprovided cresylite interfraternal
rehabilitative mediateness swearingly mericarp Protestantize drome japanned
laryngic nebular snare undiffusive insatiately nonlustrous inexistency kenno
eurythermal pyrocatechol aquiline liberatress Zuludom pelf inductivity sawdust
yeelaman cyanoguanidine antalgol redescend glacierist toplike
triakistetrahedral depressingly choralcelo dithery retinize unanatomized
aneurism hackneyed ell parastas angiolymphoma spir [...]
+unexplicit ladhood spiciferous agglomeratic veterinarian incomprehensible
allotropic sportswomanship abscission planosubulate tramplike undeterring
focaloid chilblain scrubbed furacious cromlech alen inexistency pyroacetic
selectivity pomiferous flippantness bestill wingable depressingly
sportswomanship subirrigate arrendation papery generalizable adatom ungreat
Confervales chronist collegian appetible corbel benzoperoxide experientialist
cattimandoo parquet ten nummi unleavened ascitic [...]
+pseudoxanthine pyroacetic macropterous venialness valvulotomy pictorially
Tsonecan smokefarthings Pishquow temporomastoid semiangle kenno tomorrowness
Macraucheniidae enation spookdom returnability ultrasystematic stronghearted
ovoviviparous umangite groundneedle bromate orchiocatabasis untongued throbless
seminonflammable flatman leucophoenicite bogydom Prosobranchiata raphis
inventurous Florissant collegian mutter strammel squit Bertat dunkadoo
devilwise mastication flippantness unlaps [...]
+metapolitics tendomucoid gelatinousness ultraobscure doubtingness Pyrales
isopelletierin imprescribable subdentate poleax seeingness acidophile oversand
instructiveness Caphtor electrotechnics provedore bozal tautness porencephalous
angiolymphoma Vaishnavism ovopyriform kerykeion subdrainage Auriculariales
prescriber yote knob nigh agglomeratic unbashfulness bozal almud blightbird
tendomucoid goodwill lampyrine strammel chalcites folious pseudoxanthine
discipular drome sturdied quad lien [...]
+steprelationship triakistetrahedral nonmanufacture neurodegenerative saguran
cheesecutter aconitine devilwise nativeness manilla intuition noncrystallized
warlike sheepskin asparaginic psychofugal amplexifoliate infrastapedial
outguess larklike weism chrysochrous cretaceous impressor stereotypography
helpless cattimandoo uncombable Florissant Munnopsidae parquet Inger expiscate
genii imperceptivity topline Tamil antiscolic bespin depthwise pyroacetic Mesua
bubble Animalivora flippantness [...]
+culm exprobratory signifier halloo shallowish Yannigan angiopathy
thermoresistant hymnic Bushongo roughcast Babylonism prepavement bismuthiferous
zanyism goladar phoenicochroite mesymnion wherefrom laryngic pentafid Italical
subfoliar larklike hysterogen scabbiness wemless hypoid Whilkut physiologian
lineamental manganosiderite various ventricous ethnocracy waird chilblain
chargeably taver uncarefully mutter mesymnion orthopedical rehabilitative
oinomancy metapolitics pomiferous Lentibul [...]
+ordinant uncompromisingly strander shallowish detractive cuproiodargyrite
Macraucheniidae euphemious unpeople tramplike admissory imaginary enhedge
penult fossilism totaquina beadroll skyshine perculsive Pithecolobium
Christianopaganism cubit unpredict incomprehensible Spencerism Yannigan
charioteer sterilely unsupercilious opacousness mangonism arval champer
mechanist genii monstership undeterring pelvimetry steprelationship engrain
Saponaria prescriptible plugger tristich botchedly Tam [...]
+apocalypst nummi Uraniidae unlapsing trabecular cattimandoo corona
uninhabitedness hoove floatability euphonym overcrown temporomastoid Dictograph
spiranthic unurban paranephros scotching cyanoguanidine nigh periclitation
opacousness cloy chacona twinling squdge haply inventurous cinque biodynamics
slait scapuloradial corelysis proboscidiform patroller parquet hysterogen
farrantly concretion trip beadroll diopside Endomycetaceae thermanesthesia
cacuminal undercolored drome genii toplike [...]
+Bassaris topline depthwise barkometer Passiflorales oblongly unprovided
cervisial signifier tetchy sportswomanship Dodecatheon yeat spermaphyte wandoo
crystallographical superindifference dosseret roughcast nonprofession
antineuritic debellator lebensraum gunshop Bishareen returnability Munnopsidae
shallowish arsenide stapedius sertularian overinstruct Sebastian weism
impressor pneumonalgia Hu subdentate perfunctory liberatress Mesua Hu
opacousness Inger paranephros tomorn immatchable sw [...]
+diplomatize toxihaemia bladderwort preagitate bozal pelvimetry hellbender
interruptedness pachydermatoid nigh epidymides charioteer Vaishnavism
columniform idiotize transcorporeal phallaceous rehabilitative Fouquieria
intuition valvula molecule benthonic metapolitics Fameuse porencephalous
acidophile tetchy biodynamics paranephros arrendation testa scabbardless
various nigh bettermost cockal stewardship uniarticular tonsure naught jirble
Animalivora uloid ticktick steprelationship impugn [...]
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java
b/storm-client/src/jvm/org/apache/storm/Config.java
index 79142a9c2..8d3fc51cc 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -1485,6 +1485,27 @@ public class Config extends HashMap<String, Object> {
*/
@IsString
public static final String STORM_META_SERIALIZATION_DELEGATE =
"storm.meta.serialization.delegate";
+ /**
+ * Topology configuration to enable compression of serialized tuples
during inter-worker network transfer.
+ * When set to {@code true}, tuples emitted by this component will be
compressed prior to being sent
+ * over the network to remote worker processes. This is highly recommended
for topologies exchanging
+ * large payloads (e.g., entire lines of text or large data blocks) to
significantly reduce network I/O.
+ * Default: {@code false} (Disabled by default to prevent unexpected CPU
overhead).
+ */
+ @IsBoolean
+ public static final String TOPOLOGY_TUPLE_COMPRESSION_ENABLE =
"topology.tuple.compression.enable";
+ /**
+ * Topology configuration specifying the minimum size threshold (in bytes)
for compressing serialized tuples
+ * during inter-worker network transfer.
+ * When the serialized byte array of a tuple exceeds this threshold, it
will be compressed
+ * prior to being transmitted over the network to a remote worker process.
This optimizes network I/O
+ * for large payloads (such as text blocks or massive objects) without
wasting cycles on small data.
+ * Set to 0 to bypass the size check and compress all tuples regardless of
size.
+ * Default: {@code 1460} bytes (The typical maximum segment size [MSS] for
a standard
+ * Ethernet TCP payload, preventing compression on packets that already
fit within a single network frame).
+ */
+ @IsPositiveNumber(includeZero = true)
+ public static final String TOPOLOGY_TUPLE_COMPRESSION_THRESHOLD =
"topology.tuple.compression.threshold";
/**
* GZIP max decompression bytes. Defaults to 104857600 (100MB).
*/
@@ -1503,6 +1524,11 @@ public class Config extends HashMap<String, Object> {
*/
@IsPositiveNumber(includeZero = false)
public static final String STORM_COMPRESSION_ZSTD_MAX_DECOMPRESSED_BYTES =
"storm.compression.zstd.max.decompressed.bytes";
+ /**
+ * Max decompression bytes for tuples. Defaults to 10485760 (10MB).
+ */
+ @IsPositiveNumber(includeZero = false)
+ public static final String
TOPOLOGY_TUPLE_COMPRESSION_MAX_DECOMPRESSED_BYTES =
"topology.tuple.compression.max.decompressed.bytes";
/**
* Configure the topology metrics reporters to be used on workers.
*/
diff --git
a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
index 0d9f6b9fe..a310eac9c 100644
---
a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
+++
b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
@@ -16,27 +16,65 @@ import com.esotericsoftware.kryo.io.Input;
import java.io.IOException;
import java.util.List;
import java.util.Map;
+import org.apache.storm.Config;
+import org.apache.storm.generated.ComponentCommon;
import org.apache.storm.task.GeneralTopologyContext;
import org.apache.storm.tuple.MessageId;
import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.utils.ObjectReader;
+import org.apache.storm.utils.Utils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
public class KryoTupleDeserializer implements ITupleDeserializer {
- private GeneralTopologyContext context;
- private KryoValuesDeserializer kryo;
- private SerializationFactory.IdDictionary ids;
- private Input kryoInput;
+ private static final Integer DEFAULT_MAX_DECOMPRESSED_BYTES = 10 * 1024 *
1024; // 10MBytes
+ public static final Logger LOG =
LoggerFactory.getLogger(KryoTupleDeserializer.class);
+ public static final String FAILED_TO_DESERIALIZE_TUPLE = "Failed to
deserialize tuple";
+ private final GeneralTopologyContext context;
+ private final KryoValuesDeserializer kryo;
+ private final SerializationFactory.IdDictionary ids;
+ private final Input kryoInput;
+ private final int maxZstdDecompressedBytes;
+ private final boolean anyTupleCompressionEnabled;
public KryoTupleDeserializer(final Map<String, Object> conf, final
GeneralTopologyContext context) {
kryo = new KryoValuesDeserializer(conf);
this.context = context;
ids = new SerializationFactory.IdDictionary(context.getRawTopology());
kryoInput = new Input(1);
+ maxZstdDecompressedBytes =
ObjectReader.getInt(conf.get(Config.TOPOLOGY_TUPLE_COMPRESSION_MAX_DECOMPRESSED_BYTES),
+ DEFAULT_MAX_DECOMPRESSED_BYTES);
+ anyTupleCompressionEnabled = isTupleCompressionEnabled(conf, context);
}
@Override
public TupleImpl deserialize(byte[] ser) {
+ // check zstd header if at least one component is compressing tuples.
+ if (anyTupleCompressionEnabled && Utils.ZstdUtils.isZstd(ser)) {
+ try {
+ byte[] decompressed = Utils.ZstdUtils.decompress(ser,
this.maxZstdDecompressedBytes);
+ return deserializeTuple(decompressed);
+ } catch (RuntimeException e) {
+ if (e.getMessage() != null &&
e.getMessage().contains(FAILED_TO_DESERIALIZE_TUPLE)) {
+ // isZstd() false positive: a raw Kryo tuple's first 4
bytes matched ZSTD_MAGIC_HEADER by chance.
+ // This is astronomically unlikely in practice. Because
ZSTD_MAGIC_HEADER (0xFD2FB528) is little-endian
+ // on the wire, the first byte checked is 0x28. A Kryo
writeInt(taskId, true) of 40 yields exactly 0x28.
+ // The collision is prevented not by the taskId range, but
by the second field (streamId),
+ // which would rigidly have to equal 6069 to match the
remaining magic bytes.
+ // Branch retained for correctness in case of an
accidental collision.
+ LOG.debug("isZstd() false positive: raw Kryo tuple matched
ZSTD_MAGIC_HEADER (0xFD2FB528).");
+ return deserializeTuple(ser);
+ } else {
+ throw e;
+ }
+ }
+ }
+ return deserializeTuple(ser);
+ }
+
+ private TupleImpl deserializeTuple(byte[] data) {
try {
- kryoInput.setBuffer(ser);
+ kryoInput.setBuffer(data, 0, data.length);
int taskId = kryoInput.readInt(true);
int streamId = kryoInput.readInt(true);
String componentName = context.getComponentId(taskId);
@@ -45,7 +83,21 @@ public class KryoTupleDeserializer implements
ITupleDeserializer {
List<Object> values = kryo.deserializeFrom(kryoInput);
return new TupleImpl(context, values, componentName, taskId,
streamName, id);
} catch (IOException e) {
- throw new RuntimeException(e);
+ throw new RuntimeException(FAILED_TO_DESERIALIZE_TUPLE, e);
+ }
+ }
+
+ private static boolean isTupleCompressionEnabled(final Map<String, Object>
conf, final GeneralTopologyContext context) {
+ if
(ObjectReader.getBoolean(conf.get(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE),
false)) {
+ return true;
+ }
+ for (String componentId : context.getComponentIds()) {
+ ComponentCommon common = context.getComponentCommon(componentId);
+ Map<String, Object> componentConf =
Utils.parseJson(common.get_json_conf());
+ if
(ObjectReader.getBoolean(componentConf.get(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE),
false)) {
+ return true;
+ }
}
+ return false;
}
}
diff --git
a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java
b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java
index 3b6a9f15a..7691faf06 100644
---
a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java
+++
b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleSerializer.java
@@ -14,19 +14,32 @@ package org.apache.storm.serialization;
import com.esotericsoftware.kryo.io.Output;
import java.io.IOException;
+import java.util.Arrays;
import java.util.Map;
+import org.apache.storm.Config;
import org.apache.storm.task.GeneralTopologyContext;
import org.apache.storm.tuple.Tuple;
+import org.apache.storm.utils.ObjectReader;
+import org.apache.storm.utils.Utils;
public class KryoTupleSerializer implements ITupleSerializer {
- private KryoValuesSerializer kryo;
- private SerializationFactory.IdDictionary ids;
- private Output kryoOut;
+ private static final int DEFAULT_COMPRESSION_THRESHOLD = 1460;
+ private static final Integer DEFAULT_ZSTD_COMPRESSION_LEVEL = 3;
+
+ private final KryoValuesSerializer kryo;
+ private final SerializationFactory.IdDictionary ids;
+ private final Output kryoOut;
+ private final boolean isCompressionEnabled;
+ private final int compressionThreshold;
+ private final int zstdCompressionLevel;
public KryoTupleSerializer(final Map<String, Object> conf, final
GeneralTopologyContext context) {
kryo = new KryoValuesSerializer(conf);
kryoOut = new Output(2000, 2000000000);
ids = new SerializationFactory.IdDictionary(context.getRawTopology());
+ isCompressionEnabled =
ObjectReader.getBoolean(conf.get(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE),
false);
+ compressionThreshold =
ObjectReader.getInt(conf.get(Config.TOPOLOGY_TUPLE_COMPRESSION_THRESHOLD),
DEFAULT_COMPRESSION_THRESHOLD);
+ zstdCompressionLevel =
ObjectReader.getInt(conf.get(Config.STORM_COMPRESSION_ZSTD_LEVEL),
DEFAULT_ZSTD_COMPRESSION_LEVEL);
}
@Override
@@ -38,7 +51,15 @@ public class KryoTupleSerializer implements ITupleSerializer
{
kryoOut.writeInt(ids.getStreamId(tuple.getSourceComponent(),
tuple.getSourceStreamId()), true);
tuple.getMessageId().serialize(kryoOut);
kryo.serializeInto(tuple.getValues(), kryoOut);
- return kryoOut.toBytes();
+
+ byte[] rawBytes = kryoOut.getBuffer();
+ int dataLength = kryoOut.position();
+
+ if (this.isCompressionEnabled && dataLength >
this.compressionThreshold) {
+ return Utils.ZstdUtils.compress(rawBytes, 0, dataLength,
this.zstdCompressionLevel);
+ } else {
+ return Arrays.copyOf(rawBytes, dataLength);
+ }
} catch (IOException e) {
throw new RuntimeException(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 3590583ad..e57e7f3f0 100644
--- a/storm-client/src/jvm/org/apache/storm/utils/Utils.java
+++ b/storm-client/src/jvm/org/apache/storm/utils/Utils.java
@@ -61,6 +61,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.NavigableMap;
+import java.util.Objects;
import java.util.Set;
import java.util.Stack;
import java.util.TreeMap;
@@ -1044,28 +1045,39 @@ public class Utils {
}
/**
- * Compresses the provided byte array using Zstandard.
+ * Compresses a slice of 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 offset the start offset of the slice to compress.
+ * @param length the number of bytes to compress starting at {@code
offset}.
* @param compressionLevel the zstd compression level.
- * @return a compressed byte array, or the original array if
null/empty.
+ * @return a compressed byte array, or an empty array if {@code data}
is null or {@code length} is 0.
+ * @throws IndexOutOfBoundsException if {@code offset} and {@code
length} describe a slice
+ * that falls outside the bounds of
{@code data}.
* @throws RuntimeException wrapping an {@link IOException} if the
compression fails.
*/
- public static byte[] compress(byte[] data, int compressionLevel) {
+ public static byte[] compress(byte[] data, int offset, int length, int
compressionLevel) {
if (data == null || data.length == 0) {
return data;
}
- try (ByteArrayOutputStream bos = new
ByteArrayOutputStream(data.length)) {
+ if (length == 0) {
+ return new byte[0];
+ }
+
+ // Validate the slice up front so we fail clearly before opening
any streams.
+ Objects.checkFromIndexSize(offset, length, data.length);
+
+ try (ByteArrayOutputStream bos = new
ByteArrayOutputStream(length)) {
try (ZstdCompressorOutputStream zstdOut =
ZstdCompressorOutputStream.builder()
.setOutputStream(bos)
.setBufferSize(BUFFER_SIZE) // impacts on compression
ratio
.setLevel(compressionLevel)
.get()) {
- zstdOut.write(data);
+ zstdOut.write(data, offset, length); // Write the slice
directly
zstdOut.finish();
}
return bos.toByteArray();
@@ -1074,6 +1086,24 @@ public class Utils {
}
}
+ /**
+ * 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;
+ }
+ return compress(data, 0, data.length, compressionLevel);
+ }
+
/**
* Decompresses a Zstandard-compressed byte array.
*
diff --git
a/storm-client/test/jvm/org/apache/storm/serialization/KryoTupleSerializerDeserializerTest.java
b/storm-client/test/jvm/org/apache/storm/serialization/KryoTupleSerializerDeserializerTest.java
new file mode 100644
index 000000000..bbd036970
--- /dev/null
+++
b/storm-client/test/jvm/org/apache/storm/serialization/KryoTupleSerializerDeserializerTest.java
@@ -0,0 +1,373 @@
+/**
+ * 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.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import org.apache.storm.Config;
+import org.apache.storm.generated.ComponentCommon;
+import org.apache.storm.generated.StormTopology;
+import org.apache.storm.shade.net.minidev.json.JSONValue;
+import org.apache.storm.task.GeneralTopologyContext;
+import org.apache.storm.testing.TestWordCounter;
+import org.apache.storm.testing.TestWordSpout;
+import org.apache.storm.topology.TopologyBuilder;
+import org.apache.storm.tuple.Fields;
+import org.apache.storm.tuple.MessageId;
+import org.apache.storm.tuple.TupleImpl;
+import org.apache.storm.tuple.Values;
+import org.apache.storm.utils.Utils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.MockedStatic;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.when;
+
+/**
+ * Unit tests for {@link KryoTupleSerializer} and {@link
KryoTupleDeserializer}, covering the compressed and
+ * uncompressed code paths, round-trip fidelity, mixing both encodings through
a single (de)serializer instance,
+ * and the negative/error paths.
+ */
+public class KryoTupleSerializerDeserializerTest {
+
+ private static final String SOURCE_COMPONENT = "1";
+ private static final String DEST_COMPONENT = "2";
+ private static final int SOURCE_TASK_ID = 1;
+
+ private GeneralTopologyContext context;
+
+ @BeforeEach
+ public void setup() {
+ StormTopology topology = createStormTopology();
+ context = mock(GeneralTopologyContext.class);
+ when(context.getRawTopology()).thenReturn(topology);
+
when(context.getComponentId(SOURCE_TASK_ID)).thenReturn(SOURCE_COMPONENT);
+ when(context.doSanityCheck()).thenReturn(false);
+ }
+
+ private StormTopology createStormTopology() {
+ TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout(SOURCE_COMPONENT, new TestWordSpout(true), 1);
+ builder.setBolt(DEST_COMPONENT, new TestWordCounter(),
1).fieldsGrouping(SOURCE_COMPONENT, new Fields("word"));
+ return builder.createTopology();
+ }
+
+ private Map<String, Object> baseConf() {
+ return new HashMap<>(Utils.readStormConfig());
+ }
+
+ private Map<String, Object> compressionEnabledConf(int threshold) {
+ Map<String, Object> conf = baseConf();
+ conf.put(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE, true);
+ conf.put(Config.TOPOLOGY_TUPLE_COMPRESSION_THRESHOLD, threshold);
+ conf.put(Config.STORM_COMPRESSION_ZSTD_LEVEL, 3);
+ return conf;
+ }
+
+ private void enableComponentLevelCompression(String componentId) {
+ enableComponentLevelCompression(context, componentId);
+ }
+
+ private void enableComponentLevelCompression(GeneralTopologyContext ctx,
String componentId) {
+ Map<String, Object> componentConf = new HashMap<>();
+ componentConf.put(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE, true);
+ ComponentCommon common = mock(ComponentCommon.class);
+
when(common.get_json_conf()).thenReturn(JSONValue.toJSONString(componentConf));
+
when(ctx.getComponentIds()).thenReturn(Collections.singleton(componentId));
+ when(ctx.getComponentCommon(componentId)).thenReturn(common);
+ }
+
+ private GeneralTopologyContext newContext() {
+ GeneralTopologyContext ctx = mock(GeneralTopologyContext.class);
+ when(ctx.getRawTopology()).thenReturn(createStormTopology());
+ when(ctx.getComponentId(SOURCE_TASK_ID)).thenReturn(SOURCE_COMPONENT);
+ when(ctx.doSanityCheck()).thenReturn(false);
+ return ctx;
+ }
+
+ @Test
+ public void testRoundTripUncompressedWhenCompressionDisabled() {
+ Map<String, Object> conf = baseConf(); // compression disabled by
default
+ KryoTupleSerializer serializer = new KryoTupleSerializer(conf,
context);
+ KryoTupleDeserializer deserializer = new KryoTupleDeserializer(conf,
context);
+
+ TupleImpl original = tuple(new Values("hello", 42, bigString(8192)),
MessageId.makeRootId(7L, 99L));
+ byte[] bytes = serializer.serialize(original);
+
+ assertFalse(Utils.ZstdUtils.isZstd(bytes), "compression disabled must
never emit a zstd frame");
+ assertSameTuple(original, deserializer.deserialize(bytes));
+ }
+
+ @Test
+ public void testRoundTripUncompressedWhenBelowThreshold() {
+ Map<String, Object> conf = compressionEnabledConf(1460);
+ KryoTupleSerializer serializer = new KryoTupleSerializer(conf,
context);
+ KryoTupleDeserializer deserializer = new KryoTupleDeserializer(conf,
context);
+
+ TupleImpl original = tuple(new Values("small", 1),
MessageId.makeUnanchored());
+ byte[] bytes = serializer.serialize(original);
+
+ assertFalse(Utils.ZstdUtils.isZstd(bytes), "payload below threshold
must not be compressed");
+ assertSameTuple(original, deserializer.deserialize(bytes));
+ }
+
+ @Test
+ public void testRoundTripCompressedWhenAboveThreshold() {
+ Map<String, Object> conf = compressionEnabledConf(1460);
+ KryoTupleSerializer serializer = new KryoTupleSerializer(conf,
context);
+ KryoTupleDeserializer deserializer = new KryoTupleDeserializer(conf,
context);
+
+ TupleImpl original = tuple(new Values(bigString(64 * 1024)),
MessageId.makeRootId(5L, 11L));
+ byte[] bytes = serializer.serialize(original);
+
+ assertTrue(Utils.ZstdUtils.isZstd(bytes), "payload above threshold
must be compressed");
+ assertSameTuple(original, deserializer.deserialize(bytes));
+ }
+
+ @Test
+ public void testCompressionAtExactThreshold() {
+ Map<String, Object> conf = compressionEnabledConf(0);
+ KryoTupleSerializer serializer = new KryoTupleSerializer(conf,
context);
+ KryoTupleDeserializer deserializer = new KryoTupleDeserializer(conf,
context);
+
+ TupleImpl original = tuple(new Values("x"),
MessageId.makeUnanchored());
+ byte[] bytes = serializer.serialize(original);
+
+ assertTrue(Utils.ZstdUtils.isZstd(bytes), "threshold 0 must compress
every non-empty payload");
+ assertSameTuple(original, deserializer.deserialize(bytes));
+ }
+
+ @Test
+ public void testMixedCompressedAndUncompressedSameInstances() {
+ Map<String, Object> conf = compressionEnabledConf(1460);
+ KryoTupleSerializer serializer = new KryoTupleSerializer(conf,
context);
+ KryoTupleDeserializer deserializer = new KryoTupleDeserializer(conf,
context);
+
+ TupleImpl small = tuple(new Values("tiny", 1),
MessageId.makeRootId(1L, 2L));
+ TupleImpl large = tuple(new Values(bigString(32 * 1024), "tail"),
MessageId.makeRootId(3L, 4L));
+
+ // Buffer reuse across compressed and uncompressed payloads.
+ byte[] smallBytes1 = serializer.serialize(small);
+ byte[] largeBytes1 = serializer.serialize(large);
+ byte[] smallBytes2 = serializer.serialize(small);
+ byte[] largeBytes2 = serializer.serialize(large);
+
+ assertFalse(Utils.ZstdUtils.isZstd(smallBytes1));
+ assertTrue(Utils.ZstdUtils.isZstd(largeBytes1));
+ assertFalse(Utils.ZstdUtils.isZstd(smallBytes2));
+ assertTrue(Utils.ZstdUtils.isZstd(largeBytes2));
+
+ // deserializer must transparently handle both encodings, in any order.
+ assertSameTuple(large, deserializer.deserialize(largeBytes1));
+ assertSameTuple(small, deserializer.deserialize(smallBytes1));
+ assertSameTuple(small, deserializer.deserialize(smallBytes2));
+ assertSameTuple(large, deserializer.deserialize(largeBytes2));
+ }
+
+ @Test
+ public void testComponentLevelCompressionEnablesDecompressPath() {
+ enableComponentLevelCompression(SOURCE_COMPONENT);
+ KryoTupleSerializer serializer = new
KryoTupleSerializer(compressionEnabledConf(0), context);
+ KryoTupleDeserializer deserializer = new
KryoTupleDeserializer(baseConf(), context);
+
+ TupleImpl original = tuple(new Values(bigString(16 * 1024)),
MessageId.makeRootId(8L, 9L));
+ byte[] bytes = serializer.serialize(original);
+
+ assertTrue(Utils.ZstdUtils.isZstd(bytes));
+ assertSameTuple(original, deserializer.deserialize(bytes));
+ }
+
+ @Test
+ public void testDecompressPathGatedPerTopology() {
+ // Two distinct topologies sharing the exact same compressed frame on
the wire. The gating decision is
+ // per-topology (workers are per-topology), so the same bytes must be
decompressed by one and not the other.
+ GeneralTopologyContext compressingTopology = newContext();
+ enableComponentLevelCompression(compressingTopology,
SOURCE_COMPONENT); // at least one component compresses
+ GeneralTopologyContext plainTopology = newContext();
// no component enables compression
+
+ KryoTupleSerializer serializer = new
KryoTupleSerializer(compressionEnabledConf(0), compressingTopology);
+ TupleImpl original = tuple(new Values(bigString(16 * 1024)),
MessageId.makeRootId(8L, 9L));
+ byte[] bytes = serializer.serialize(original);
+ assertTrue(Utils.ZstdUtils.isZstd(bytes), "precondition: serializer
produced a compressed frame");
+
+ KryoTupleDeserializer compressingDeser = new
KryoTupleDeserializer(baseConf(), compressingTopology);
+ KryoTupleDeserializer plainDeser = new
KryoTupleDeserializer(baseConf(), plainTopology);
+
+ // Compressing topology: decompress path is taken, the frame
round-trips.
+ assertSameTuple(original, compressingDeser.deserialize(bytes));
+ // Plain topology: decompress path is skipped, the frame is treated as
a raw kryo tuple and fails to parse.
+ assertThrows(RuntimeException.class, () ->
plainDeser.deserialize(bytes));
+ }
+
+ @Test
+ public void testNoComponentCompressionSkipsDecompressPath() {
+ KryoTupleSerializer serializer = new
KryoTupleSerializer(compressionEnabledConf(0), context);
+ KryoTupleDeserializer deserializer = new
KryoTupleDeserializer(baseConf(), context);
+
+ byte[] bytes = serializer.serialize(tuple(new Values(bigString(16 *
1024)), MessageId.makeRootId(8L, 9L)));
+ assertTrue(Utils.ZstdUtils.isZstd(bytes), "precondition: serializer
produced a compressed frame");
+ assertThrows(RuntimeException.class, () ->
deserializer.deserialize(bytes));
+ }
+
+ @Test
+ public void testDecompressFailureFallsBackToRawTupleParsing() {
+ // Exercises the false-positive fallback in
KryoTupleDeserializer#deserialize: compression is enabled and
+ // isZstd() reports a match, but the decompress path surfaces a
"Failed to deserialize tuple" error.
+ enableComponentLevelCompression(SOURCE_COMPONENT); //
anyTupleCompressionEnabled == true
+ KryoTupleSerializer serializer = new KryoTupleSerializer(baseConf(),
context);
+ KryoTupleDeserializer deserializer = new
KryoTupleDeserializer(baseConf(), context);
+
+ TupleImpl original = tuple(new Values("hello", 42),
MessageId.makeRootId(5L, 11L));
+ byte[] raw = serializer.serialize(original);
+ assertFalse(Utils.ZstdUtils.isZstd(raw), "precondition: serializer
produced a raw, uncompressed tuple");
+
+ try (MockedStatic<Utils.ZstdUtils> mocked =
mockStatic(Utils.ZstdUtils.class)) {
+ // Force entry into the decompress branch, then make decompression
report a tuple-deserialization failure.
+ mocked.when(() -> Utils.ZstdUtils.isZstd(raw)).thenReturn(true);
+ mocked.when(() -> Utils.ZstdUtils.decompress(eq(raw), anyInt()))
+ .thenThrow(new
RuntimeException(KryoTupleDeserializer.FAILED_TO_DESERIALIZE_TUPLE));
+
+ TupleImpl result = deserializer.deserialize(raw);
+ assertSameTuple(original, result); // fallback re-parsed the raw
bytes and recovered the tuple
+ }
+ }
+
+ @Test
+ public void testDecompressFailureFallbackRethrowsWhenRawBytesAlsoInvalid()
{
+ // the raw bytes are not a valid kryo tuple either: the fallback parse
+ // must surface a RuntimeException rather than silently returning a
bogus tuple.
+ enableComponentLevelCompression(SOURCE_COMPONENT);
+ KryoTupleDeserializer deserializer = new
KryoTupleDeserializer(baseConf(), context);
+
+ byte[] raw = {1, 2, 3, 4, 5, 6, 7, 8};
+ try (MockedStatic<Utils.ZstdUtils> mocked =
mockStatic(Utils.ZstdUtils.class)) {
+ mocked.when(() -> Utils.ZstdUtils.isZstd(raw)).thenReturn(true);
+ mocked.when(() -> Utils.ZstdUtils.decompress(eq(raw), anyInt()))
+ .thenThrow(new
RuntimeException(KryoTupleDeserializer.FAILED_TO_DESERIALIZE_TUPLE));
+
+ assertThrows(RuntimeException.class, () ->
deserializer.deserialize(raw));
+ }
+ }
+
+ // corner cases
+
+ @Test
+ public void testRoundTripEmptyValues() {
+ Map<String, Object> conf = baseConf();
+ KryoTupleSerializer serializer = new KryoTupleSerializer(conf,
context);
+ KryoTupleDeserializer deserializer = new KryoTupleDeserializer(conf,
context);
+
+ TupleImpl original = tuple(new Values(), MessageId.makeUnanchored());
+ TupleImpl result =
deserializer.deserialize(serializer.serialize(original));
+ assertSameTuple(original, result);
+ assertTrue(result.getValues().isEmpty());
+ }
+
+ @Test
+ public void testRoundTripNullValue() {
+ Map<String, Object> conf = baseConf();
+ KryoTupleSerializer serializer = new KryoTupleSerializer(conf,
context);
+ KryoTupleDeserializer deserializer = new KryoTupleDeserializer(conf,
context);
+
+ TupleImpl original = tuple(new Values("a", null, "b"),
MessageId.makeUnanchored());
+ assertSameTuple(original,
deserializer.deserialize(serializer.serialize(original)));
+ }
+
+ // negative tests
+
+ @Test
+ public void testDeserializeGarbageBytesThrows() {
+ KryoTupleDeserializer deserializer = new
KryoTupleDeserializer(baseConf(), context);
+ // Not a zstd frame (length < 4 keeps isZstd false) and not a valid
kryo tuple.
+ assertThrows(RuntimeException.class, () ->
deserializer.deserialize(new byte[]{1, 2, 3}));
+ }
+
+ @Test
+ public void testDeserializeEmptyBytesThrows() {
+ KryoTupleDeserializer deserializer = new
KryoTupleDeserializer(baseConf(), context);
+ assertThrows(RuntimeException.class, () ->
deserializer.deserialize(new byte[0]));
+ }
+
+ @Test
+ public void testDeserializeZstdMagicButInvalidFrameThrows() {
+ enableComponentLevelCompression(SOURCE_COMPONENT);
+ KryoTupleDeserializer deserializer = new
KryoTupleDeserializer(baseConf(), context);
+ // First 4 bytes are the little-endian zstd magic 0xFD2FB528 so
isZstd() is true, but the rest is
+ // not a valid zstd frame. decompress() throws, the false-positive
fallback re-parses the raw bytes,
+ // which are also not a valid kryo tuple -> RuntimeException.
+ byte[] fakeZstd = new byte[]{(byte) 0x28, (byte) 0xB5, (byte) 0x2F,
(byte) 0xFD, 0x00, 0x01, 0x02, 0x03};
+ assertTrue(Utils.ZstdUtils.isZstd(fakeZstd));
+ assertThrows(RuntimeException.class, () ->
deserializer.deserialize(fakeZstd));
+ }
+
+ @Test
+ public void testDeserializeCompressedExceedingMaxDecompressedBytesThrows()
{
+ // A genuinely compressed tuple, but the deserializer is configured
with a tiny decompression cap.
+ // decompress() throws ("threshold exceeded"), the fallback re-parses
the still-compressed raw bytes,
+ // which are not a valid kryo tuple -> RuntimeException.
+ enableComponentLevelCompression(SOURCE_COMPONENT);
+ KryoTupleSerializer serializer = new
KryoTupleSerializer(compressionEnabledConf(0), context);
+ Map<String, Object> deserConf = baseConf();
+
deserConf.put(Config.TOPOLOGY_TUPLE_COMPRESSION_MAX_DECOMPRESSED_BYTES, 1);
+ KryoTupleDeserializer deserializer = new
KryoTupleDeserializer(deserConf, context);
+
+ byte[] bytes = serializer.serialize(tuple(new Values(bigString(8192)),
MessageId.makeUnanchored()));
+ assertTrue(Utils.ZstdUtils.isZstd(bytes));
+ assertThrows(RuntimeException.class, () ->
deserializer.deserialize(bytes));
+ }
+
+ @Test
+ public void testSerializeUnregisteredTypeWithoutJavaFallbackThrows() {
+ Map<String, Object> conf = baseConf();
+ conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, false);
+ KryoTupleSerializer serializer = new KryoTupleSerializer(conf,
context);
+
+ TupleImpl original = tuple(new Values(new UnregisteredType()),
MessageId.makeUnanchored());
+ assertThrows(RuntimeException.class, () ->
serializer.serialize(original));
+ }
+
+ private static class UnregisteredType {
+ @SuppressWarnings("unused")
+ private final int field = 1;
+ }
+
+ private TupleImpl tuple(List<Object> values, MessageId id) {
+ return new TupleImpl(context, values, SOURCE_COMPONENT,
SOURCE_TASK_ID, Utils.DEFAULT_STREAM_ID, id);
+ }
+
+ private void assertSameTuple(TupleImpl expected, TupleImpl actual) {
+ assertEquals(expected.getValues(), actual.getValues());
+ assertEquals(expected.getSourceTask(), actual.getSourceTask());
+ assertEquals(expected.getSourceComponent(),
actual.getSourceComponent());
+ assertEquals(expected.getSourceStreamId(), actual.getSourceStreamId());
+ assertEquals(expected.getMessageId(), actual.getMessageId());
+ }
+
+ private String bigString(int size) {
+ StringBuilder sb = new StringBuilder(size);
+ for (int i = 0; i < size; i++) {
+ sb.append((char) ('a' + (i % 26)));
+ }
+ return sb.toString();
+ }
+}
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 d29f41d50..6086f5095 100644
--- a/storm-client/test/jvm/org/apache/storm/utils/UtilsTest.java
+++ b/storm-client/test/jvm/org/apache/storm/utils/UtilsTest.java
@@ -566,6 +566,56 @@ public class UtilsTest {
}
}
+ @Test
+ void compress_offsetLength_compressesOnlyTheSlice() {
+ byte[] prefix = "PREFIX-".getBytes(StandardCharsets.UTF_8);
+ byte[] payload = "Hello, ZSTD!".getBytes(StandardCharsets.UTF_8);
+ byte[] suffix = "-SUFFIX".getBytes(StandardCharsets.UTF_8);
+
+ byte[] framed = new byte[prefix.length + payload.length +
suffix.length];
+ System.arraycopy(prefix, 0, framed, 0, prefix.length);
+ System.arraycopy(payload, 0, framed, prefix.length, payload.length);
+ System.arraycopy(suffix, 0, framed, prefix.length + payload.length,
suffix.length);
+
+ byte[] compressedSlice = Utils.ZstdUtils.compress(framed,
prefix.length, payload.length, 3);
+ byte[] decompressed = Utils.ZstdUtils.decompress(compressedSlice, 1024
* 1024);
+ assertArrayEquals(payload, decompressed,
+ "Compressing a slice then decompressing must recover only that
slice");
+
+ // The slice path must match the full-array path when the slice spans
the whole array.
+ byte[] compressedWhole = Utils.ZstdUtils.compress(payload, 3);
+ byte[] compressedFullSlice = Utils.ZstdUtils.compress(payload, 0,
payload.length, 3);
+ assertArrayEquals(compressedWhole, compressedFullSlice,
+ "Whole-array slice must produce the same output as the
convenience overload");
+ }
+
+ @Test
+ void compress_offsetLength_zeroLengthOrNull() {
+ byte[] empty = Utils.ZstdUtils.compress(new byte[]{1, 2, 3}, 1, 0, 3);
+ assertNotNull(empty);
+ assertEquals(0, empty.length);
+
+ byte[] fromNull = Utils.ZstdUtils.compress(null, 0, 0, 3);
+ assertNull(fromNull);
+
+ byte[] emptyArray = new byte[0];
+ byte[] fromEmpty = Utils.ZstdUtils.compress(emptyArray, 0, 0, 3);
+ assertEquals(emptyArray, fromEmpty);
+ }
+
+ @Test
+ void compress_offsetLength_outOfBounds_throws() {
+ byte[] data = {1, 2, 3, 4};
+ assertThrows(IndexOutOfBoundsException.class,
+ () -> Utils.ZstdUtils.compress(data, -1, 2, 3));
+ assertThrows(IndexOutOfBoundsException.class,
+ () -> Utils.ZstdUtils.compress(data, 0, -1, 3));
+ assertThrows(IndexOutOfBoundsException.class,
+ () -> Utils.ZstdUtils.compress(data, 3, 2, 3));
+ assertThrows(IndexOutOfBoundsException.class,
+ () -> Utils.ZstdUtils.compress(data, 5, 1, 3));
+ }
+
@Test
void compress_highEntropData_doesNotThrow() {
// Random-ish bytes — Zstd may not shrink them, but must not fail
