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

Author: L1nq0

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"

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

@@ -1648,6 +1648,15 @@ 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.
+     * 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,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);
+    }
+}

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<String, Object> 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<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 testJavaFallbackMissingClassDroppedAndBatchContinues() {
         Map<String, Object> conf = baseConf();