Fix NPE in WorkerState.transferLocalBatch causing worker crashes

storm-client/src/jvm/org/apache/storm/daemon/worker/WorkerState.java

@@ -566,10 +566,17 @@ public boolean tryFlushRemotes() {
 
     // Receives msgs from remote workers and feeds them to local executors. If any receiving local executor is under Back Pressure,
     // informs other workers about back pressure situation. Runs in the NettyWorker thread.
-    private void transferLocalBatch(ArrayList<AddressedTuple> tupleBatch) {
+    // Package-private for testing.
+    void transferLocalBatch(ArrayList<AddressedTuple> tupleBatch) {
         for (int i = 0; i < tupleBatch.size(); i++) {
             AddressedTuple tuple = tupleBatch.get(i);
             JCQueue queue = taskToExecutorQueue.get(tuple.dest);
+            if (queue == null) {
+                // tuple addressed to a task not assigned to this worker, e.g. sent by a peer
+                // holding a stale assignment during a rebalance (STORM-3751)
+                dropMessage(tuple);
+                continue;
+            }
 
             // 1- try adding to main queue if its overflow is not empty
             if (queue.isEmptyOverflow()) {
@@ -609,6 +616,12 @@ private void dropMessage(AddressedTuple tuple, JCQueue queue) {
             queue.getQueueName(), queue.getOverflowCount(), dropCount, tuple);
     }
 
+    private void dropMessage(AddressedTuple tuple) {
+        ++dropCount;
+        LOG.warn("Dropping message for unknown task {}. Total Drop Count = {}, Dropped Message : {}",
+            tuple.dest, dropCount, tuple);
+    }
+
     public void checkSerialize(KryoTupleSerializer serializer, AddressedTuple tuple) {
         if (trySerializeLocal) {
             serializer.serialize(tuple.getTuple());

storm-client/test/jvm/org/apache/storm/daemon/worker/WorkerStateTest.java

@@ -20,17 +20,21 @@
 import org.apache.storm.daemon.supervisor.AdvancedFSOps;
 import org.apache.storm.generated.StormTopology;
 import org.apache.storm.thrift.TException;
+import org.apache.storm.tuple.AddressedTuple;
+import org.apache.storm.tuple.Tuple;
 import org.apache.storm.utils.ConfigUtils;
 import org.apache.storm.utils.Utils;
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNull;
@@ -107,4 +111,31 @@ public void testVisibilityOfUserResource() throws IOException, TException {
             ConfigUtils.setInstance(previousConfigUtils);
         }
     }
+
+    @Test
+    public void testTransferLocalBatchDropsTuplesForUnknownTasks() throws TException, IOException {
+        ConfigUtils mockedConfigUtils = mock(ConfigUtils.class);
+        ConfigUtils previousConfigUtils = ConfigUtils.setInstance(mockedConfigUtils);
+
+        try {
+            Map<String, Object> conf = new HashMap<>();
+            conf.put(Config.TOPOLOGY_WORKER_SHARED_THREAD_POOL_SIZE, 1);
+
+            String topologyId = "1";
+            StormTopology topology = mock(StormTopology.class);
+            when(topology.deepCopy()).thenReturn(topology);
+            when(mockedConfigUtils.readSupervisorTopologyImpl(eq(conf), eq(topologyId), any(AdvancedFSOps.class))).thenReturn(topology);
+
+            WorkerState workerState = TestUtilsForWorkerState.getWorkerState(conf, topologyId);
+
+            // the mocked assignment contains no executors, so no task id is known to this worker;
+            // a remote peer with a stale assignment may still address tuples to it (STORM-3751)
+            AddressedTuple tupleForUnknownTask = new AddressedTuple(42, mock(Tuple.class));
+
+            assertDoesNotThrow(() ->
+                workerState.transferLocalBatch(new ArrayList<>(Collections.singletonList(tupleForUnknownTask))));
+        } finally {
+            ConfigUtils.setInstance(previousConfigUtils);
+        }
+    }
 }