You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@storm.apache.org by ra...@apache.org on 2026/05/30 15:32:04 UTC
(storm) branch master updated: Add tuple compression for inter-worker communication (#8707)
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 <48...@users.noreply.github.com>
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