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 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)
 

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. 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
 

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

@@ -1648,6 +1648,19 @@ 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. 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
+    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

@@ -43,7 +43,8 @@ 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(
         IOException.class,
         KryoException.class,
@@ -71,6 +72,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 +90,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 +108,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/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<Integer, String> streamNames = streamIdToName.get(component);
+            return streamNames == null ? null : streamNames.get(stream);
         }
     }
 }

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 IllegalArgumentException {
+    public TupleDeserializationException(String message) {
+        super(message);
+    }
+}

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,62 @@ 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();
+        // 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(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(undeclaredStreamId, true);
+        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 +318,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;

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<String, Object> conf) {