Drop malformed tuple payloads instead of killing the receiving worker

docs/Metrics.md

@@ -295,12 +295,14 @@ Be aware that the `__system` bolt is an actual bolt so regular bolt metrics desc
     "dequeuedMessages": 0,
     "enqueued": {
       "/127.0.0.1:49952": 389951
-    }
+    },
+    "deserializationFailures": 0
 }
 ```
 
 `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.
 
 ##### Send (Netty Client)
 

storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java

@@ -12,10 +12,16 @@
 
 package org.apache.storm.messaging;
 
+import com.esotericsoftware.kryo.KryoException;
+import java.io.IOException;
+import java.nio.BufferUnderflowException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
 import org.apache.storm.Config;
@@ -26,17 +32,41 @@
 import org.apache.storm.tuple.AddressedTuple;
 import org.apache.storm.tuple.Tuple;
 import org.apache.storm.utils.ObjectReader;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 
 /**
  * A class that is called when a TaskMessage arrives.
  */
 public class DeserializingConnectionCallback implements IConnectionCallback, IMetric {
+    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.
+    private static final Set<Class<?>> TOLERATED_DESERIALIZATION_FAILURES = new HashSet<>(Arrays.asList(
+        IOException.class,
+        KryoException.class,
+        IllegalArgumentException.class,
+        NegativeArraySizeException.class,
+        ClassCastException.class,
+        ArrayIndexOutOfBoundsException.class,
+        BufferUnderflowException.class,
+        ClassNotFoundException.class));
+
+    // Drop-log rate limits: the first INDIVIDUAL_DROP_LOG_LIMIT failures are logged in full,
+    // then one summary line per DROP_LOG_SUMMARY_INTERVAL further failures, and a run of
+    // CONSECUTIVE_DROP_WARN_THRESHOLD failures without a successful deserialization logs
+    // a single WARN.
+    private static final int INDIVIDUAL_DROP_LOG_LIMIT = 10;
+    private static final int DROP_LOG_SUMMARY_INTERVAL = 100;
+    private static final long CONSECUTIVE_DROP_WARN_THRESHOLD = 1000L;
+
     private final WorkerState.ILocalTransferCallback cb;
     private final Map<String, Object> conf;
     private final GeneralTopologyContext context;
 
-    private final ThreadLocal<KryoTupleDeserializer> des =
+    private ThreadLocal<KryoTupleDeserializer> des =
         new ThreadLocal<KryoTupleDeserializer>() {
             @Override
             protected KryoTupleDeserializer initialValue() {
@@ -47,7 +77,12 @@ protected KryoTupleDeserializer initialValue() {
     // Track serialized size of messages.
     private final boolean sizeMetricsEnabled;
     private final ConcurrentHashMap<String, AtomicLong> byteCounts = new ConcurrentHashMap<>();
+    private final AtomicLong deserializationFailures = new AtomicLong(0L);
 
+    // Log-limit counters are separate from the deserializationFailures metric: metric reads
+    // reset that counter, which would restart the limits.
+    private final AtomicLong totalDropCount = new AtomicLong(0L);
+    private final AtomicLong consecutiveDropCount = new AtomicLong(0L);
 
     public DeserializingConnectionCallback(final Map<String, Object> conf, final GeneralTopologyContext context,
                                            WorkerState.ILocalTransferCallback callback) {
@@ -58,23 +93,65 @@ public DeserializingConnectionCallback(final Map<String, Object> conf, final Gen
 
     }
 
+    // Package-private for testing.
+    void setDeserializer(KryoTupleDeserializer replacement) {
+        this.des = ThreadLocal.withInitial(() -> replacement);
+    }
+
     @Override
     public void recv(List<TaskMessage> batch) {
         KryoTupleDeserializer des = this.des.get();
         ArrayList<AddressedTuple> ret = new ArrayList<>(batch.size());
         for (TaskMessage message : batch) {
-            Tuple tuple = des.deserialize(message.message());
-            AddressedTuple addrTuple = new AddressedTuple(message.task(), tuple);
-            updateMetrics(tuple.getSourceTask(), message);
-            ret.add(addrTuple);
+            try {
+                Tuple tuple = des.deserialize(message.message());
+                AddressedTuple addrTuple = new AddressedTuple(message.task(), tuple);
+                updateMetrics(tuple.getSourceTask(), message);
+                ret.add(addrTuple);
+                if (consecutiveDropCount.get() != 0L) {
+                    consecutiveDropCount.set(0L);
+                }
+            } catch (Exception e) {
+                if (!isToleratedDeserializationFailure(e)) {
+                    throw e;
+                }
+                deserializationFailures.incrementAndGet();
+                long totalDrops = totalDropCount.incrementAndGet();
+                if (totalDrops <= INDIVIDUAL_DROP_LOG_LIMIT) {
+                    LOG.error("Failed to deserialize a message of {} bytes destined for task {}, dropping it",
+                              message.message().length, message.task(), e);
+                } else if ((totalDrops - INDIVIDUAL_DROP_LOG_LIMIT)
+                           % DROP_LOG_SUMMARY_INTERVAL == 0) {
+                    LOG.error("Dropped {} further messages that failed to deserialize "
+                              + "since the last summary. Total Drop Count= {}",
+                              DROP_LOG_SUMMARY_INTERVAL, totalDrops, e);
+                }
+                long consecutiveDrops = consecutiveDropCount.incrementAndGet();
+                if (consecutiveDrops == CONSECUTIVE_DROP_WARN_THRESHOLD) {
+                    LOG.warn("{} consecutive messages have failed to deserialize, indicating "
+                             + "a persistent fault such as a class missing from the classpath",
+                             consecutiveDrops, e);
+                }
+            }
         }
         cb.transfer(ret);
     }
 
+    private static boolean isToleratedDeserializationFailure(Exception e) {
+        for (Throwable t = e; t != null; t = t.getCause()) {
+            for (Class<?> klass : TOLERATED_DESERIALIZATION_FAILURES) {
+                if (klass.isInstance(t)) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
     /**
      * Returns serialized byte count traffic metrics.
      *
-     * @return Map of metric counts, or null if disabled
+     * @return Map of metric counts, or null when size metrics are disabled
      */
     @Override
     public Object getValueAndReset() {
@@ -91,6 +168,14 @@ public Object getValueAndReset() {
         return outMap;
     }
 
+    /**
+     * Returns the number of messages dropped because deserialization failed since the last call,
+     * and resets the count.
+     */
+    public long getAndResetDeserializationFailures() {
+        return deserializationFailures.getAndSet(0L);
+    }
+
     /**
      * Update serialized byte counts for each message.
      *

storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java

@@ -26,6 +26,7 @@
 import org.apache.storm.Config;
 import org.apache.storm.grouping.Load;
 import org.apache.storm.messaging.ConnectionWithStatus;
+import org.apache.storm.messaging.DeserializingConnectionCallback;
 import org.apache.storm.messaging.IConnectionCallback;
 import org.apache.storm.messaging.TaskMessage;
 import org.apache.storm.metric.api.IMetric;
@@ -241,6 +242,11 @@ public Object getState() {
         }
         ret.put("enqueued", enqueued);
 
+        if (cb instanceof DeserializingConnectionCallback) {
+            DeserializingConnectionCallback callback = (DeserializingConnectionCallback) cb;
+            ret.put("deserializationFailures", callback.getAndResetDeserializationFailures());
+        }
+
         // Report messageSizes metric, if enabled (non-null).
         if (cb instanceof IMetric) {
             Object metrics = ((IMetric) cb).getValueAndReset();

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

@@ -78,6 +78,9 @@ private TupleImpl deserializeTuple(byte[] data) {
             int taskId = kryoInput.readInt(true);
             int streamId = kryoInput.readInt(true);
             String componentName = context.getComponentId(taskId);
+            if (componentName == null) {
+                throw new IllegalArgumentException("Received a tuple from unknown task " + taskId);
+            }
             String streamName = ids.getStreamName(componentName, streamId);
             MessageId id = MessageId.deserialize(kryoInput);
             List<Object> values = kryo.deserializeFrom(kryoInput);