Throw a typed exception for tuples with unknown task or stream ids and add a deserialization strict mode

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"

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)
 

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
 

storm-client/src/jvm/org/apache/storm/Config.java

@@ -1648,6 +1648,16 @@ public class Config extends HashMap<String, Object> {
      */
     @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. 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
+    public static final String TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE = "topology.tuple.deserialization.strict.enable";
     /**
      * Configure the topology metrics reporters to be used on workers.
      */

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<Class<?>> 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<String, AtomicLong> byteCounts = new ConcurrentHashMap<>();
@@ -87,6 +92,7 @@ public DeserializingConnectionCallback(final Map<String, Object> 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<TaskMessage> batch) {
             try {
                 tuple = des.deserialize(message.message());
             } catch (Exception e) {
-                if (!isToleratedDeserializationFailure(e)) {
+                if (strictMode || !isToleratedDeserializationFailure(e)) {
                     throw e;
                 }
                 deserializationFailures.incrementAndGet();

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<Object> values = kryo.deserializeFrom(kryoInput);
             return new TupleImpl(context, values, componentName, taskId, streamName, id);

storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java

@@ -0,0 +1,26 @@
+/**
+ * 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 IllegalArgumentException {
+    public TupleDeserializationException(String message) {
+        super(message);
+    }
+
+    public TupleDeserializationException(String message, Throwable cause) {
+        super(message, cause);
+    }
+}

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;
@@ -132,14 +133,60 @@ public void testTruncatedKryoPayloadDroppedAndBatchContinues() {
     @Test
     public void testUnknownSourceTaskDroppedAndBatchContinues() {
         Map<String, Object> conf = baseConf();
+
+        TupleDeserializationException thrown = assertThrows(TupleDeserializationException.class,
+                                                            () -> new KryoTupleDeserializer(conf, context).deserialize(unknownSourceTaskTuple()));
+        assertTrue(thrown.getMessage().contains("9999"),
+                   "expected the task id in the message but was: " + thrown.getMessage());
+
+        assertBatchDeliversOnlyValidMessages(conf, unknownSourceTaskTuple());
+    }
+
+    @Test
+    public void testUnknownStreamIdDroppedAndBatchContinues() {
+        Map<String, Object> 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();
+        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();
 
-        assertThrows(IllegalArgumentException.class, () -> new KryoTupleDeserializer(conf, context).deserialize(unknownTask));
+        TupleDeserializationException thrown = assertThrows(TupleDeserializationException.class,
+                                                            () -> new KryoTupleDeserializer(conf, context).deserialize(unknownStream));
+        assertTrue(thrown.getMessage().contains(SOURCE_COMPONENT),
+                   "expected the component name in the message but was: " + thrown.getMessage());
 
-        assertBatchDeliversOnlyValidMessages(conf, unknownTask);
+        assertBatchDeliversOnlyValidMessages(conf, unknownStream);
+    }
+
+    @Test
+    public void testStrictModeMakesFailuresFatal() {
+        Map<String, Object> 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 testStrictModeMakesUnknownTaskFailureFatal() {
+        Map<String, Object> 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
@@ -269,6 +316,13 @@ private void assertBatchDeliversOnlyValidMessages(Map<String, Object> 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<String, Object> baseConf() {
         Map<String, Object> conf = new HashMap<>(Utils.readStormConfig());
         return conf;