From 7654c7b1ec469dba060349736dfe3e1f2117a6a3 Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Wed, 16 Sep 2026 10:47:34 +0800 Subject: [PATCH 1/2] Throw a typed exception for tuples with unknown task or stream ids and add a deserialization strict mode --- conf/defaults.yaml | 1 + .../src/jvm/org/apache/storm/Config.java | 9 +++++ .../DeserializingConnectionCallback.java | 10 ++++- .../serialization/KryoTupleDeserializer.java | 5 ++- .../TupleDeserializationException.java | 22 +++++++++++ .../DeserializingConnectionCallbackTest.java | 38 ++++++++++++++++++- 6 files changed, 81 insertions(+), 4 deletions(-) create mode 100644 storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java diff --git a/conf/defaults.yaml b/conf/defaults.yaml index 6fd7a04b9d..939d79b1e5 100644 --- a/conf/defaults.yaml +++ b/conf/defaults.yaml @@ -61,6 +61,7 @@ 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 +topology.tuple.deserialization.strict.enable: false storm.codedistributor.class: "org.apache.storm.codedistributor.LocalFileSystemCodeDistributor" storm.workers.artifacts.dir: "workers-artifacts" storm.health.check.dir: "healthchecks" diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index 519b8d943b..d8dd4ebc4c 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -1648,6 +1648,15 @@ public class Config extends HashMap { */ @IsPositiveNumber(includeZero = false) public static final String TOPOLOGY_TUPLE_COMPRESSION_MAX_DECOMPRESSED_BYTES = "topology.tuple.compression.max.decompressed.bytes"; + /** + * Topology configuration to make tuple deserialization failures fatal instead of dropping the undecodable message. + * By default a message that fails to decode on the receiving worker is dropped and counted, and the worker keeps + * running. When set to {@code true}, any deserialization failure propagates and the worker exits, restoring the + * pre-3.1.0 behavior. + * Default: {@code false}. + */ + @IsBoolean + public static final String TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE = "topology.tuple.deserialization.strict.enable"; /** * Configure the topology metrics reporters to be used on workers. */ diff --git a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java index 6a8464f432..37aaf1bec5 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java @@ -28,6 +28,7 @@ import org.apache.storm.daemon.worker.WorkerState; import org.apache.storm.metric.api.IMetric; import org.apache.storm.serialization.KryoTupleDeserializer; +import org.apache.storm.serialization.TupleDeserializationException; import org.apache.storm.task.GeneralTopologyContext; import org.apache.storm.tuple.AddressedTuple; import org.apache.storm.tuple.Tuple; @@ -43,8 +44,10 @@ public class DeserializingConnectionCallback implements IConnectionCallback, IMe private static final Logger LOG = LoggerFactory.getLogger(DeserializingConnectionCallback.class); // A tuple that cannot be decoded is dropped instead of killing the worker; anything outside this set keeps - // the fatal handling in StormServerHandler. + // the fatal handling in StormServerHandler. TupleDeserializationException is thrown by KryoTupleDeserializer + // for unknown task or stream ids. private static final Set> TOLERATED_DESERIALIZATION_FAILURES = new HashSet<>(Arrays.asList( + TupleDeserializationException.class, IOException.class, KryoException.class, IllegalArgumentException.class, @@ -71,6 +74,8 @@ protected KryoTupleDeserializer initialValue() { } }; + private final boolean strictMode; + // Track serialized size of messages. private final boolean sizeMetricsEnabled; private final ConcurrentHashMap byteCounts = new ConcurrentHashMap<>(); @@ -87,6 +92,7 @@ public DeserializingConnectionCallback(final Map conf, final Gen this.context = context; cb = callback; sizeMetricsEnabled = ObjectReader.getBoolean(conf.get(Config.TOPOLOGY_SERIALIZED_MESSAGE_SIZE_METRICS), false); + strictMode = ObjectReader.getBoolean(conf.get(Config.TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE), false); } @@ -104,7 +110,7 @@ public void recv(List batch) { try { tuple = des.deserialize(message.message()); } catch (Exception e) { - if (!isToleratedDeserializationFailure(e)) { + if (strictMode || !isToleratedDeserializationFailure(e)) { throw e; } deserializationFailures.incrementAndGet(); 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 301a8ca96d..e433603f18 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java @@ -79,9 +79,12 @@ private TupleImpl deserializeTuple(byte[] data) { int streamId = kryoInput.readInt(true); String componentName = context.getComponentId(taskId); if (componentName == null) { - throw new IllegalArgumentException("Received a tuple from unknown task " + taskId); + throw new TupleDeserializationException("Received a tuple from unknown task " + taskId); } String streamName = ids.getStreamName(componentName, streamId); + if (streamName == null) { + throw new TupleDeserializationException("Component " + componentName + " has no stream with id " + streamId); + } MessageId id = MessageId.deserialize(kryoInput); List values = kryo.deserializeFrom(kryoInput); return new TupleImpl(context, values, componentName, taskId, streamName, id); diff --git a/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java b/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java new file mode 100644 index 0000000000..9ebb38f7c9 --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java @@ -0,0 +1,22 @@ +/** + * 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; + +/** + * Thrown when a serialized tuple names a source task or stream that the receiving topology cannot resolve. + */ +public class TupleDeserializationException extends RuntimeException { + public TupleDeserializationException(String message) { + super(message); + } +} diff --git a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java index 2a982b3a3b..a22aba4f9d 100644 --- a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java +++ b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java @@ -27,6 +27,7 @@ import org.apache.storm.daemon.worker.WorkerState; import org.apache.storm.serialization.KryoTupleDeserializer; import org.apache.storm.serialization.KryoTupleSerializer; +import org.apache.storm.serialization.TupleDeserializationException; import org.apache.storm.task.GeneralTopologyContext; import org.apache.storm.testing.TestWordCounter; import org.apache.storm.testing.TestWordSpout; @@ -137,11 +138,46 @@ public void testUnknownSourceTaskDroppedAndBatchContinues() { out.writeInt(1, true); // default stream id byte[] unknownTask = out.toBytes(); - assertThrows(IllegalArgumentException.class, () -> new KryoTupleDeserializer(conf, context).deserialize(unknownTask)); + TupleDeserializationException thrown = assertThrows(TupleDeserializationException.class, + () -> new KryoTupleDeserializer(conf, context).deserialize(unknownTask)); + assertTrue(thrown.getMessage().contains("9999"), + "expected the task id in the message but was: " + thrown.getMessage()); assertBatchDeliversOnlyValidMessages(conf, unknownTask); } + @Test + public void testUnknownStreamIdDroppedAndBatchContinues() { + Map conf = baseConf(); + Output out = new Output(16, 32); + out.writeInt(SOURCE_TASK_ID, true); // source task that exists in the topology + out.writeInt(3, true); // stream id the source component does not declare + byte[] unknownStream = out.toBytes(); + + TupleDeserializationException thrown = assertThrows(TupleDeserializationException.class, + () -> new KryoTupleDeserializer(conf, context).deserialize(unknownStream)); + assertTrue(thrown.getMessage().contains("id 3"), + "expected the stream id in the message but was: " + thrown.getMessage()); + + assertBatchDeliversOnlyValidMessages(conf, unknownStream); + } + + @Test + public void testStrictModeMakesFailuresFatal() { + Map conf = baseConf(); + conf.put(Config.TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE, true); + byte[] full = serializedTuple(conf, new Values("a-string-long-enough-to-survive-truncation", 7)); + byte[] truncated = Arrays.copyOf(full, full.length - 10); + + WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); + DeserializingConnectionCallback callback = new DeserializingConnectionCallback(conf, context, transfer); + + assertThrows(KryoException.class, () -> callback.recv(Collections.singletonList(taskMessage(truncated)))); + + verify(transfer, never()).transfer(any()); + assertEquals(0L, callback.getAndResetDeserializationFailures()); + } + @Test public void testJavaFallbackMissingClassDroppedAndBatchContinues() { Map conf = baseConf(); From be82785185a51e31f5e435e2edff215949b80be4 Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Wed, 16 Sep 2026 16:40:04 +0800 Subject: [PATCH 2/2] Make TupleDeserializationException extend IllegalArgumentException and document the strict mode The exception thrown for unknown task or stream ids now extends IllegalArgumentException, so handlers written against 3.1.0 keep matching, and it gains a (String, Throwable) constructor. The strict mode flag is documented in Serialization.md and Metrics.md, and the config javadoc spells out that a persistent bad frame puts the worker in a restart loop. The unknown-stream test no longer pins the exact message text and the strict mode is now covered for the typed exception as well. --- docs/Metrics.md | 2 +- docs/Serialization.md | 1 + .../src/jvm/org/apache/storm/Config.java | 3 +- .../TupleDeserializationException.java | 6 +++- .../DeserializingConnectionCallbackTest.java | 34 ++++++++++++++----- 5 files changed, 35 insertions(+), 11 deletions(-) diff --git a/docs/Metrics.md b/docs/Metrics.md index e7349f7718..d79c3c6ec1 100644 --- a/docs/Metrics.md +++ b/docs/Metrics.md @@ -302,7 +302,7 @@ Be aware that the `__system` bolt is an actual bolt so regular bolt metrics desc `dequeuedMessages` is a throwback to older code where there was an internal queue between the server and the bolts/spouts. That is no longer the case and the value can be ignored. `enqueued` is a map between the address of the remote worker and the number of tuples that were sent from it to this worker. -`deserializationFailures` is the number of incoming messages that failed to deserialize and were dropped. +`deserializationFailures` is the number of incoming messages that failed to deserialize and were dropped. When `topology.tuple.deserialization.strict.enable` is set, deserialization failures are not dropped or counted; they propagate and terminate the worker instead. ##### Send (Netty Client) diff --git a/docs/Serialization.md b/docs/Serialization.md index b8701e8aac..7cdaeabc1d 100644 --- a/docs/Serialization.md +++ b/docs/Serialization.md @@ -131,6 +131,7 @@ Be aware that the topology-wide form enables compression for *every* remote-boun | `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. | +| `topology.tuple.deserialization.strict.enable` | `false` | Makes tuple deserialization failures fatal: an incoming message that fails to decode kills the receiving worker instead of being dropped and counted. This restores the pre-3.1.0 behavior and is mainly useful for debugging, because a persistent bad frame will put the worker in a restart loop. | #### How decompression works diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index d8dd4ebc4c..986e84f676 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -1652,7 +1652,8 @@ public class Config extends HashMap { * Topology configuration to make tuple deserialization failures fatal instead of dropping the undecodable message. * By default a message that fails to decode on the receiving worker is dropped and counted, and the worker keeps * running. When set to {@code true}, any deserialization failure propagates and the worker exits, restoring the - * pre-3.1.0 behavior. + * pre-3.1.0 behavior. Be aware that a single corrupt frame from a peer then kills the worker, and the supervisor + * restarts it into the same failure, so a persistent bad frame results in a restart loop. * Default: {@code false}. */ @IsBoolean diff --git a/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java b/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java index 9ebb38f7c9..188a90b9e4 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java @@ -15,8 +15,12 @@ /** * Thrown when a serialized tuple names a source task or stream that the receiving topology cannot resolve. */ -public class TupleDeserializationException extends RuntimeException { +public class TupleDeserializationException extends IllegalArgumentException { public TupleDeserializationException(String message) { super(message); } + + public TupleDeserializationException(String message, Throwable cause) { + super(message, cause); + } } diff --git a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java index a22aba4f9d..4fd9cbdb09 100644 --- a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java +++ b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java @@ -133,17 +133,13 @@ public void testTruncatedKryoPayloadDroppedAndBatchContinues() { @Test public void testUnknownSourceTaskDroppedAndBatchContinues() { Map conf = baseConf(); - Output out = new Output(16, 32); - out.writeInt(9999, true); // source task that does not exist in the topology - out.writeInt(1, true); // default stream id - byte[] unknownTask = out.toBytes(); TupleDeserializationException thrown = assertThrows(TupleDeserializationException.class, - () -> new KryoTupleDeserializer(conf, context).deserialize(unknownTask)); + () -> new KryoTupleDeserializer(conf, context).deserialize(unknownSourceTaskTuple())); assertTrue(thrown.getMessage().contains("9999"), "expected the task id in the message but was: " + thrown.getMessage()); - assertBatchDeliversOnlyValidMessages(conf, unknownTask); + assertBatchDeliversOnlyValidMessages(conf, unknownSourceTaskTuple()); } @Test @@ -156,8 +152,8 @@ public void testUnknownStreamIdDroppedAndBatchContinues() { TupleDeserializationException thrown = assertThrows(TupleDeserializationException.class, () -> new KryoTupleDeserializer(conf, context).deserialize(unknownStream)); - assertTrue(thrown.getMessage().contains("id 3"), - "expected the stream id in the message but was: " + thrown.getMessage()); + assertTrue(thrown.getMessage().contains(SOURCE_COMPONENT), + "expected the component name in the message but was: " + thrown.getMessage()); assertBatchDeliversOnlyValidMessages(conf, unknownStream); } @@ -178,6 +174,21 @@ public void testStrictModeMakesFailuresFatal() { assertEquals(0L, callback.getAndResetDeserializationFailures()); } + @Test + public void testStrictModeMakesUnknownTaskFailureFatal() { + Map conf = baseConf(); + conf.put(Config.TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE, true); + + WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); + DeserializingConnectionCallback callback = new DeserializingConnectionCallback(conf, context, transfer); + + assertThrows(TupleDeserializationException.class, + () -> callback.recv(Collections.singletonList(taskMessage(unknownSourceTaskTuple())))); + + verify(transfer, never()).transfer(any()); + assertEquals(0L, callback.getAndResetDeserializationFailures()); + } + @Test public void testJavaFallbackMissingClassDroppedAndBatchContinues() { Map conf = baseConf(); @@ -305,6 +316,13 @@ private void assertBatchDeliversOnlyValidMessages(Map conf, byte assertNull(callback.getValueAndReset()); } + private static byte[] unknownSourceTaskTuple() { + Output out = new Output(16, 32); + out.writeInt(9999, true); // source task that does not exist in the topology + out.writeInt(1, true); // default stream id + return out.toBytes(); + } + private Map baseConf() { Map conf = new HashMap<>(Utils.readStormConfig()); return conf;