From 6ab2adc6104a938c4750bb5e9cfb44cd4f164844 Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Wed, 16 Sep 2026 10:47:34 +0800 Subject: [PATCH 1/3] 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 6fd7a04b9d4..939d79b1e5c 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 519b8d943bb..d8dd4ebc4c4 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 6a8464f432c..37aaf1bec58 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 301a8ca96de..e433603f18f 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 00000000000..9ebb38f7c95 --- /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 2a982b3a3b6..a22aba4f9d2 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 1e4e849243ebf5f64cd5c0fed9e4188661b08924 Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Wed, 16 Sep 2026 16:40:04 +0800 Subject: [PATCH 2/3] 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 e7349f77187..d79c3c6ec1b 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 b8701e8aacc..7cdaeabc1dd 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 d8dd4ebc4c4..986e84f6766 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 9ebb38f7c95..188a90b9e4d 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 a22aba4f9d2..4fd9cbdb095 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; From c3d3747c10028a5ed6971639ca0b20f6d493c17e Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Tue, 6 Oct 2026 11:24:25 +0800 Subject: [PATCH 3/3] Return null for unknown components in IdDictionary and document the strict-mode boundary for IOException-wrapped failures The tolerated-failures set matches TupleDeserializationException through its IllegalArgumentException supertype, so the explicit entry and the unused two-argument constructor are removed. IdDictionary.getStreamName returns null when the component is unknown, so both unresolved-routing cases reach the typed exception instead of a NullPointerException that stays fatal in default mode. The strict-mode documentation now separates the two outcomes: decode failures exit the worker, while failures wrapping an IOException close the connection and drop the batch with the worker kept running. The unknown-stream test derives its id from the declared stream count instead of a hardcoded constant. --- docs/Metrics.md | 2 +- docs/Serialization.md | 2 +- storm-client/src/jvm/org/apache/storm/Config.java | 5 ++++- .../messaging/DeserializingConnectionCallback.java | 2 -- .../storm/serialization/SerializationFactory.java | 3 ++- .../serialization/TupleDeserializationException.java | 4 ---- .../DeserializingConnectionCallbackTest.java | 4 +++- .../serialization/SerializationFactoryTest.java | 12 ++++++++++++ 8 files changed, 23 insertions(+), 11 deletions(-) diff --git a/docs/Metrics.md b/docs/Metrics.md index d79c3c6ec1b..9a3a9fc83aa 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. When `topology.tuple.deserialization.strict.enable` is set, deserialization failures are not dropped or counted; they propagate and terminate the worker instead. +`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 instead. Non-IO failures terminate the worker, while failures caused by an `IOException`, such as a corrupt compressed frame, close the connection and leave the worker running. ##### Send (Netty Client) diff --git a/docs/Serialization.md b/docs/Serialization.md index 7cdaeabc1dd..4d3841ad896 100644 --- a/docs/Serialization.md +++ b/docs/Serialization.md @@ -131,7 +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. | +| `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. A failure rooted in an `IOException`, such as a corrupt or oversized compressed frame, is an exception: the receiving worker closes that connection, discards the batch it was reading, and carries on while the sender reconnects. | #### 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 986e84f6766..9c275f82c40 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -1653,7 +1653,10 @@ public class Config extends HashMap { * 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. 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. + * restarts it into the same failure, so a persistent bad frame results in a restart loop. Failures that wrap an + * {@code IOException}, such as a broken or oversized compressed frame, do not reach the exit path: the messaging + * layer treats them like a connection error, drops the batch being read, and leaves the worker running while the + * peer reconnects. * Default: {@code false}. */ @IsBoolean 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 37aaf1bec58..cee8776a1a4 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java @@ -28,7 +28,6 @@ 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; @@ -47,7 +46,6 @@ public class DeserializingConnectionCallback implements IConnectionCallback, IMe // 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, diff --git a/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java b/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java index 18b471c99ff..9180978ea45 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java @@ -259,7 +259,8 @@ public int getStreamId(String component, String stream) { } public String getStreamName(String component, int stream) { - return streamIdToName.get(component).get(stream); + Map streamNames = streamIdToName.get(component); + return streamNames == null ? null : streamNames.get(stream); } } } 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 188a90b9e4d..8329ea19862 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java +++ b/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java @@ -19,8 +19,4 @@ 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 4fd9cbdb095..de8bf62b291 100644 --- a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java +++ b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java @@ -145,9 +145,11 @@ public void testUnknownSourceTaskDroppedAndBatchContinues() { @Test public void testUnknownStreamIdDroppedAndBatchContinues() { Map conf = baseConf(); + // IdDictionary assigns ids 1..n to the component's n declared streams, so the first id past that count is undeclared. + int undeclaredStreamId = Utils.getComponentCommon(context.getRawTopology(), SOURCE_COMPONENT).get_streams().size() + 1; 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 + out.writeInt(undeclaredStreamId, true); byte[] unknownStream = out.toBytes(); TupleDeserializationException thrown = assertThrows(TupleDeserializationException.class, diff --git a/storm-client/test/jvm/org/apache/storm/serialization/SerializationFactoryTest.java b/storm-client/test/jvm/org/apache/storm/serialization/SerializationFactoryTest.java index d7e7bfcd43b..998098643cc 100644 --- a/storm-client/test/jvm/org/apache/storm/serialization/SerializationFactoryTest.java +++ b/storm-client/test/jvm/org/apache/storm/serialization/SerializationFactoryTest.java @@ -16,11 +16,14 @@ import java.util.Map; import org.apache.storm.Config; import org.apache.storm.security.serialization.BlowfishTupleSerializer; +import org.apache.storm.testing.TestWordSpout; +import org.apache.storm.topology.TopologyBuilder; import org.apache.storm.utils.ListDelegate; import org.apache.storm.utils.Utils; import org.junit.jupiter.api.Test; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; public class SerializationFactoryTest { @@ -65,6 +68,15 @@ public void test_registers_decorator_with_conf() { kryo.getSerializer(MockSerObject.class).getClass()); } + @Test + public void test_get_stream_name_null_for_unknown_component() { + TopologyBuilder builder = new TopologyBuilder(); + builder.setSpout("spout", new TestWordSpout(true), 1); + SerializationFactory.IdDictionary ids = new SerializationFactory.IdDictionary(builder.createTopology()); + + assertNull(ids.getStreamName("no-such-component", 1)); + } + public static class MockKryoDecorator implements IKryoDecorator { @Override public void decorate(Kryo k, Map conf) {