From 24a700e68c3e482c6ce7828668e351eddc6b45f7 Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Fri, 4 Sep 2026 16:32:19 +0800 Subject: [PATCH 1/2] Drop malformed tuple payloads instead of killing the receiving worker A tuple payload that cannot be decoded escaped recv() into the Netty fatal handler, terminating the worker. The supervisor restarted the worker, and the same poison message terminated it again. recv() now catches per-message deserialization failures whose cause chain contains one of the exceptions raised by undecodable payloads (IOException, KryoException, IllegalArgumentException, NegativeArraySizeException, ClassCastException, ArrayIndexOutOfBoundsException, BufferUnderflowException, NullPointerException, ClassNotFoundException). The offending message is dropped, the failure is logged with the destination task and payload size, the count is exposed as a deserializationFailures metric next to the message size metrics, and the rest of the batch is delivered. Any other Exception still propagates unchanged, and Errors are not caught. https://github.com/apache/storm/issues/9074 --- .../DeserializingConnectionCallback.java | 64 ++++- .../DeserializingConnectionCallbackTest.java | 218 ++++++++++++++++++ 2 files changed, 275 insertions(+), 7 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java index b038e026c2b..18048437ff5 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java @@ -12,12 +12,19 @@ package org.apache.storm.messaging; +import java.io.IOException; +import java.nio.BufferUnderflowException; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; 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 com.esotericsoftware.kryo.KryoException; import org.apache.storm.Config; import org.apache.storm.daemon.worker.WorkerState; import org.apache.storm.metric.api.IMetric; @@ -26,12 +33,32 @@ import org.apache.storm.tuple.AddressedTuple; import org.apache.storm.tuple.Tuple; import org.apache.storm.utils.ObjectReader; +import org.apache.storm.utils.Utils; +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> TOLERATED_DESERIALIZATION_FAILURES = new HashSet<>(Arrays.asList( + IOException.class, + KryoException.class, + IllegalArgumentException.class, + NegativeArraySizeException.class, + ClassCastException.class, + ArrayIndexOutOfBoundsException.class, + BufferUnderflowException.class, + NullPointerException.class, + ClassNotFoundException.class)); + + static final String DESERIALIZATION_FAILURES_KEY = "deserializationFailures"; + private final WorkerState.ILocalTransferCallback cb; private final Map conf; private final GeneralTopologyContext context; @@ -47,6 +74,7 @@ protected KryoTupleDeserializer initialValue() { // Track serialized size of messages. private final boolean sizeMetricsEnabled; private final ConcurrentHashMap byteCounts = new ConcurrentHashMap<>(); + private final AtomicLong deserializationFailures = new AtomicLong(0L); public DeserializingConnectionCallback(final Map conf, final GeneralTopologyContext context, @@ -63,23 +91,42 @@ public void recv(List batch) { KryoTupleDeserializer des = this.des.get(); ArrayList 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); + } catch (Exception e) { + if (!isToleratedDeserializationFailure(e)) { + throw e; + } + deserializationFailures.incrementAndGet(); + LOG.error("Failed to deserialize a message of {} bytes destined for task {}, dropping it", + message.message().length, message.task(), e); + } } cb.transfer(ret); } + private static boolean isToleratedDeserializationFailure(Exception e) { + for (Class klass : TOLERATED_DESERIALIZATION_FAILURES) { + if (Utils.exceptionCauseIsInstanceOf(klass, e)) { + return true; + } + } + return false; + } + /** - * Returns serialized byte count traffic metrics. + * Returns serialized byte count traffic metrics and the count of dropped deserialization failures. * - * @return Map of metric counts, or null if disabled + * @return Map of metric counts, or null when size metrics are disabled and no failures occurred */ @Override public Object getValueAndReset() { + long failures = deserializationFailures.getAndSet(0L); if (!sizeMetricsEnabled) { - return null; + return failures > 0 ? Collections.singletonMap(DESERIALIZATION_FAILURES_KEY, failures) : null; } HashMap outMap = new HashMap<>(); for (Map.Entry ent : byteCounts.entrySet()) { @@ -88,6 +135,9 @@ public Object getValueAndReset() { outMap.put(ent.getKey(), count.getAndSet(0L)); } } + if (failures > 0) { + outMap.put(DESERIALIZATION_FAILURES_KEY, failures); + } return outMap; } diff --git a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java index f622ade12ce..b85ef8ffb86 100644 --- a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java +++ b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java @@ -12,30 +12,71 @@ package org.apache.storm.messaging; +import com.esotericsoftware.kryo.KryoException; +import com.esotericsoftware.kryo.io.Output; +import java.io.IOException; +import java.io.Serializable; +import java.lang.reflect.Field; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; +import java.util.List; import java.util.Map; import org.apache.storm.Config; import org.apache.storm.daemon.worker.WorkerState; +import org.apache.storm.serialization.KryoTupleDeserializer; +import org.apache.storm.serialization.KryoTupleSerializer; import org.apache.storm.task.GeneralTopologyContext; +import org.apache.storm.testing.TestWordCounter; +import org.apache.storm.testing.TestWordSpout; +import org.apache.storm.topology.TopologyBuilder; +import org.apache.storm.tuple.AddressedTuple; +import org.apache.storm.tuple.Fields; +import org.apache.storm.tuple.MessageId; +import org.apache.storm.tuple.TupleImpl; +import org.apache.storm.tuple.Values; +import org.apache.storm.utils.Utils; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; +import org.mockito.ArgumentCaptor; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; public class DeserializingConnectionCallbackTest { private static final byte[] messageBytes = new byte[3]; private static TaskMessage message; + private static final String SOURCE_COMPONENT = "1"; + private static final String DEST_COMPONENT = "2"; + private static final int SOURCE_TASK_ID = 1; + private static final int DEST_TASK_ID = 2; + private static final byte[] JAVA_STREAM_HEADER = {(byte) 0xAC, (byte) 0xED, 0x00, 0x05}; + + private GeneralTopologyContext context; + @BeforeEach public void setUp() throws Exception { // Setup a test message message = mock(TaskMessage.class); when(message.task()).thenReturn(456); // destination taskId when(message.message()).thenReturn(messageBytes); + + TopologyBuilder builder = new TopologyBuilder(); + builder.setSpout(SOURCE_COMPONENT, new TestWordSpout(true), 1); + builder.setBolt(DEST_COMPONENT, new TestWordCounter(), 1).fieldsGrouping(SOURCE_COMPONENT, new Fields("word")); + context = mock(GeneralTopologyContext.class); + when(context.getRawTopology()).thenReturn(builder.createTopology()); + when(context.getComponentId(SOURCE_TASK_ID)).thenReturn(SOURCE_COMPONENT); } @@ -77,4 +118,181 @@ public void testUpdateMetricsConfigOn() { assertTrue(metrics instanceof Map); assertEquals(6L, ((Map) metrics).get("123-456")); } + + @Test + public void testBatchWithCorruptMessageDropsOnlyCorruptMessage() { + Map conf = baseConf(); + byte[] corrupt = new byte[]{1, 2, 3}; + assertThrows(RuntimeException.class, + () -> new KryoTupleDeserializer(conf, context).deserialize(corrupt)); + + assertBatchDeliversOnlyValidMessages(conf, corrupt); + } + + @Test + public void testTruncatedKryoPayloadDroppedAndBatchContinues() { + Map conf = baseConf(); + byte[] full = serializedTuple(conf, new Values("a-string-long-enough-to-survive-truncation", 7)); + byte[] truncated = Arrays.copyOf(full, full.length - 10); + + assertThrows(KryoException.class, () -> new KryoTupleDeserializer(conf, context).deserialize(truncated)); + + assertBatchDeliversOnlyValidMessages(conf, truncated); + } + + @Test + public void testUnknownSourceTaskDroppedAndBatchContinues() { + Map 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(); + + assertThrows(NullPointerException.class, () -> new KryoTupleDeserializer(conf, context).deserialize(unknownTask)); + + assertBatchDeliversOnlyValidMessages(conf, unknownTask); + } + + @Test + public void testJavaFallbackMissingClassDroppedAndBatchContinues() { + Map conf = baseConf(); + conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, true); + byte[] bytes = serializedTuple(conf, Collections.singletonList(new JavaSerializedValue())); + byte[] missingClass = replaceAll(bytes, "JavaSerializedValue", "JavaSerializedValuf"); + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> new KryoTupleDeserializer(conf, context).deserialize(missingClass)); + assertTrue(Utils.exceptionCauseIsInstanceOf(ClassNotFoundException.class, thrown), + "expected a ClassNotFoundException in the cause chain but was: " + thrown); + + assertBatchDeliversOnlyValidMessages(conf, missingClass); + } + + @Test + public void testJavaFallbackNegativeLengthDroppedAndBatchContinues() { + Map conf = baseConf(); + conf.put(Config.TOPOLOGY_FALL_BACK_ON_JAVA_SERIALIZATION, true); + byte[] bytes = serializedTuple(conf, Collections.singletonList(new JavaSerializedValue())); + + // SerializableSerializer writes the java-serialization byte count right before the stream header; + // an all-bits-set count makes it allocate a negative-length array. + int headerIdx = indexOf(bytes, JAVA_STREAM_HEADER, 0); + assertTrue(headerIdx >= 4, "java serialization header not found in tuple payload"); + for (int i = 1; i <= 4; i++) { + bytes[headerIdx - i] = (byte) 0xFF; + } + + assertThrows(NegativeArraySizeException.class, () -> new KryoTupleDeserializer(conf, context).deserialize(bytes)); + + assertBatchDeliversOnlyValidMessages(conf, bytes); + } + + @Test + public void testIoExceptionFailureDroppedAndBatchContinues() { + Map conf = baseConf(); + conf.put(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE, true); + byte[] fakeZstd = {(byte) 0x28, (byte) 0xB5, (byte) 0x2F, (byte) 0xFD, 0x00, 0x01, 0x02, 0x03}; + + RuntimeException thrown = assertThrows(RuntimeException.class, + () -> new KryoTupleDeserializer(conf, context).deserialize(fakeZstd.clone())); + assertTrue(Utils.exceptionCauseIsInstanceOf(IOException.class, thrown), + "expected an IOException in the cause chain but was: " + thrown); + + assertBatchDeliversOnlyValidMessages(conf, fakeZstd); + } + + @Test + public void testNonToleratedExceptionPropagates() throws Exception { + WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); + DeserializingConnectionCallback callback = new DeserializingConnectionCallback(baseConf(), context, transfer); + KryoTupleDeserializer failing = mock(KryoTupleDeserializer.class); + when(failing.deserialize(any(byte[].class))).thenThrow(new IllegalStateException("injected")); + replaceDeserializer(callback, failing); + + assertThrows(IllegalStateException.class, + () -> callback.recv(Collections.singletonList(taskMessage(new byte[]{1})))); + + verify(transfer, never()).transfer(any()); + assertNull(callback.getValueAndReset()); + } + + private static void replaceDeserializer(DeserializingConnectionCallback callback, + KryoTupleDeserializer replacement) throws Exception { + Field field = DeserializingConnectionCallback.class.getDeclaredField("des"); + field.setAccessible(true); + field.set(callback, new ThreadLocal() { + @Override + protected KryoTupleDeserializer initialValue() { + return replacement; + } + }); + } + + private void assertBatchDeliversOnlyValidMessages(Map conf, byte[] badPayload) { + WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); + DeserializingConnectionCallback callback = new DeserializingConnectionCallback(conf, context, transfer); + + callback.recv(Arrays.asList( + taskMessage(serializedTuple(conf, new Values("nathan", 1))), + taskMessage(badPayload), + taskMessage(serializedTuple(conf, new Values("golda", 2))))); + + @SuppressWarnings("unchecked") + ArgumentCaptor> captor = ArgumentCaptor.forClass(ArrayList.class); + verify(transfer).transfer(captor.capture()); + List delivered = captor.getValue(); + assertEquals(2, delivered.size()); + assertEquals(DEST_TASK_ID, delivered.get(0).getDest()); + assertEquals(new Values("nathan", 1), delivered.get(0).getTuple().getValues()); + assertEquals(DEST_TASK_ID, delivered.get(1).getDest()); + assertEquals(new Values("golda", 2), delivered.get(1).getTuple().getValues()); + + Object metrics = callback.getValueAndReset(); + assertTrue(metrics instanceof Map, "expected the deserialization failure to be counted"); + assertEquals(1L, ((Map) metrics).get(DeserializingConnectionCallback.DESERIALIZATION_FAILURES_KEY)); + assertNull(callback.getValueAndReset()); + } + + private Map baseConf() { + Map conf = new HashMap<>(Utils.readStormConfig()); + return conf; + } + + private byte[] serializedTuple(Map conf, List values) { + TupleImpl tuple = new TupleImpl(context, values, SOURCE_COMPONENT, SOURCE_TASK_ID, + Utils.DEFAULT_STREAM_ID, MessageId.makeUnanchored()); + return new KryoTupleSerializer(conf, context).serialize(tuple); + } + + private static TaskMessage taskMessage(byte[] payload) { + return new TaskMessage(DEST_TASK_ID, payload); + } + + private static byte[] replaceAll(byte[] src, String from, String to) { + byte[] out = src.clone(); + byte[] fromBytes = from.getBytes(StandardCharsets.US_ASCII); + byte[] toBytes = to.getBytes(StandardCharsets.US_ASCII); + int idx = indexOf(out, fromBytes, 0); + while (idx >= 0) { + System.arraycopy(toBytes, 0, out, idx, toBytes.length); + idx = indexOf(out, fromBytes, idx + toBytes.length); + } + return out; + } + + private static int indexOf(byte[] src, byte[] pattern, int from) { + outer: + for (int i = from; i <= src.length - pattern.length; i++) { + for (int j = 0; j < pattern.length; j++) { + if (src[i + j] != pattern[j]) { + continue outer; + } + } + return i; + } + return -1; + } + + private static class JavaSerializedValue implements Serializable { + } } From 553b9b304af686aa96dd5ad55b1d64a8d5b20298 Mon Sep 17 00:00:00 2001 From: L1nq0 Date: Sat, 5 Sep 2026 08:05:43 +0800 Subject: [PATCH 2/2] Rate-limit drop logging and report deserializationFailures from the server A topology stuck receiving poison payloads would flood the worker log with one ERROR per dropped message. recv() now logs the first 10 failures individually, then one summary ERROR per 100 further failures carrying the running total, in the WorkerState "Total Drop Count= {}" style. 1000 consecutive failures without a success log a single WARN pointing at a persistent fault; any successful deserialization resets that counter. NullPointerException stays outside the tolerated set: it usually signals a bug rather than a malformed payload. The case a bad tuple could trigger, an unknown source task, is rejected up front in KryoTupleDeserializer with IllegalArgumentException naming the task; that lookup NPEd during stream resolution before this change. Server.getState() publishes deserializationFailures as a top-level key, always present, including when it is 0, read through getAndResetDeserializationFailures() on the callback. getValueAndReset() reports only the size metrics, null when they are disabled. isToleratedDeserializationFailure walks the exception cause chain once and checks every tolerated type per frame, instead of once per type. Tests inject a replacement deserializer through a package-private setter, and a new ServerTest covers the top-level key. https://github.com/apache/storm/issues/9074 --- docs/Metrics.md | 4 +- .../DeserializingConnectionCallback.java | 71 ++++++++++++++----- .../apache/storm/messaging/netty/Server.java | 6 ++ .../serialization/KryoTupleDeserializer.java | 3 + .../DeserializingConnectionCallbackTest.java | 42 ++++++----- .../storm/messaging/netty/ServerTest.java | 43 +++++++++++ 6 files changed, 132 insertions(+), 37 deletions(-) create mode 100644 storm-client/test/jvm/org/apache/storm/messaging/netty/ServerTest.java diff --git a/docs/Metrics.md b/docs/Metrics.md index 8d620f0da9e..e7349f77187 100644 --- a/docs/Metrics.md +++ b/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) diff --git a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java index 18048437ff5..53453266b80 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java @@ -12,11 +12,11 @@ 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.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.List; @@ -24,7 +24,6 @@ import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicLong; -import com.esotericsoftware.kryo.KryoException; import org.apache.storm.Config; import org.apache.storm.daemon.worker.WorkerState; import org.apache.storm.metric.api.IMetric; @@ -33,7 +32,6 @@ import org.apache.storm.tuple.AddressedTuple; import org.apache.storm.tuple.Tuple; import org.apache.storm.utils.ObjectReader; -import org.apache.storm.utils.Utils; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -54,16 +52,21 @@ public class DeserializingConnectionCallback implements IConnectionCallback, IMe ClassCastException.class, ArrayIndexOutOfBoundsException.class, BufferUnderflowException.class, - NullPointerException.class, ClassNotFoundException.class)); - static final String DESERIALIZATION_FAILURES_KEY = "deserializationFailures"; + // 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 conf; private final GeneralTopologyContext context; - private final ThreadLocal des = + private ThreadLocal des = new ThreadLocal() { @Override protected KryoTupleDeserializer initialValue() { @@ -76,6 +79,10 @@ protected KryoTupleDeserializer initialValue() { private final ConcurrentHashMap 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 conf, final GeneralTopologyContext context, WorkerState.ILocalTransferCallback callback) { @@ -86,6 +93,11 @@ public DeserializingConnectionCallback(final Map conf, final Gen } + // Package-private for testing. + void setDeserializer(KryoTupleDeserializer replacement) { + this.des = ThreadLocal.withInitial(() -> replacement); + } + @Override public void recv(List batch) { KryoTupleDeserializer des = this.des.get(); @@ -96,37 +108,55 @@ public void recv(List batch) { 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(); - LOG.error("Failed to deserialize a message of {} bytes destined for task {}, dropping it", - message.message().length, message.task(), e); + 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 (Class klass : TOLERATED_DESERIALIZATION_FAILURES) { - if (Utils.exceptionCauseIsInstanceOf(klass, e)) { - return true; + 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 and the count of dropped deserialization failures. + * Returns serialized byte count traffic metrics. * - * @return Map of metric counts, or null when size metrics are disabled and no failures occurred + * @return Map of metric counts, or null when size metrics are disabled */ @Override public Object getValueAndReset() { - long failures = deserializationFailures.getAndSet(0L); if (!sizeMetricsEnabled) { - return failures > 0 ? Collections.singletonMap(DESERIALIZATION_FAILURES_KEY, failures) : null; + return null; } HashMap outMap = new HashMap<>(); for (Map.Entry ent : byteCounts.entrySet()) { @@ -135,12 +165,17 @@ public Object getValueAndReset() { outMap.put(ent.getKey(), count.getAndSet(0L)); } } - if (failures > 0) { - outMap.put(DESERIALIZATION_FAILURES_KEY, failures); - } 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. * diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java index da5adeacf24..3d54cab0d7d 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java +++ b/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(); diff --git a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java index a310eac9c7e..301a8ca96de 100644 --- a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java +++ b/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 values = kryo.deserializeFrom(kryoInput); diff --git a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java index b85ef8ffb86..b6c36643ced 100644 --- a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java +++ b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java @@ -16,7 +16,6 @@ import com.esotericsoftware.kryo.io.Output; import java.io.IOException; import java.io.Serializable; -import java.lang.reflect.Field; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.Arrays; @@ -148,7 +147,7 @@ public void testUnknownSourceTaskDroppedAndBatchContinues() { out.writeInt(1, true); // default stream id byte[] unknownTask = out.toBytes(); - assertThrows(NullPointerException.class, () -> new KryoTupleDeserializer(conf, context).deserialize(unknownTask)); + assertThrows(IllegalArgumentException.class, () -> new KryoTupleDeserializer(conf, context).deserialize(unknownTask)); assertBatchDeliversOnlyValidMessages(conf, unknownTask); } @@ -201,33 +200,42 @@ public void testIoExceptionFailureDroppedAndBatchContinues() { assertBatchDeliversOnlyValidMessages(conf, fakeZstd); } + @Test + public void testFailuresCountedSeparatelyFromSizeMetrics() { + Map conf = baseConf(); + conf.put(Config.TOPOLOGY_SERIALIZED_MESSAGE_SIZE_METRICS, Boolean.TRUE); + WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); + DeserializingConnectionCallback callback = new DeserializingConnectionCallback(conf, context, transfer); + + callback.recv(Arrays.asList( + taskMessage(serializedTuple(conf, new Values("nathan", 1))), + taskMessage(new byte[]{1, 2, 3}))); + + Object metrics = callback.getValueAndReset(); + assertTrue(metrics instanceof Map); + assertEquals(1, ((Map) metrics).size()); + assertTrue(((Map) metrics).containsKey("1-2")); + + assertEquals(1L, callback.getAndResetDeserializationFailures()); + assertEquals(0L, callback.getAndResetDeserializationFailures()); + } + @Test public void testNonToleratedExceptionPropagates() throws Exception { WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); DeserializingConnectionCallback callback = new DeserializingConnectionCallback(baseConf(), context, transfer); KryoTupleDeserializer failing = mock(KryoTupleDeserializer.class); when(failing.deserialize(any(byte[].class))).thenThrow(new IllegalStateException("injected")); - replaceDeserializer(callback, failing); + callback.setDeserializer(failing); assertThrows(IllegalStateException.class, () -> callback.recv(Collections.singletonList(taskMessage(new byte[]{1})))); verify(transfer, never()).transfer(any()); + assertEquals(0L, callback.getAndResetDeserializationFailures()); assertNull(callback.getValueAndReset()); } - private static void replaceDeserializer(DeserializingConnectionCallback callback, - KryoTupleDeserializer replacement) throws Exception { - Field field = DeserializingConnectionCallback.class.getDeclaredField("des"); - field.setAccessible(true); - field.set(callback, new ThreadLocal() { - @Override - protected KryoTupleDeserializer initialValue() { - return replacement; - } - }); - } - private void assertBatchDeliversOnlyValidMessages(Map conf, byte[] badPayload) { WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); DeserializingConnectionCallback callback = new DeserializingConnectionCallback(conf, context, transfer); @@ -247,9 +255,7 @@ private void assertBatchDeliversOnlyValidMessages(Map conf, byte assertEquals(DEST_TASK_ID, delivered.get(1).getDest()); assertEquals(new Values("golda", 2), delivered.get(1).getTuple().getValues()); - Object metrics = callback.getValueAndReset(); - assertTrue(metrics instanceof Map, "expected the deserialization failure to be counted"); - assertEquals(1L, ((Map) metrics).get(DeserializingConnectionCallback.DESERIALIZATION_FAILURES_KEY)); + assertEquals(1L, callback.getAndResetDeserializationFailures()); assertNull(callback.getValueAndReset()); } diff --git a/storm-client/test/jvm/org/apache/storm/messaging/netty/ServerTest.java b/storm-client/test/jvm/org/apache/storm/messaging/netty/ServerTest.java new file mode 100644 index 00000000000..f1f7ca6fe68 --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/messaging/netty/ServerTest.java @@ -0,0 +1,43 @@ +/** + * 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 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.messaging.netty; + +import java.util.Map; +import org.apache.storm.messaging.DeserializingConnectionCallback; +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.assertTrue; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class ServerTest { + + @Test + public void testGetStateReportsDeserializationFailures() { + DeserializingConnectionCallback cb = mock(DeserializingConnectionCallback.class); + when(cb.getAndResetDeserializationFailures()).thenReturn(7L, 0L); + Server server = new Server(Utils.readStormConfig(), 0, cb, null); + try { + Object state = server.getState(); + assertTrue(state instanceof Map); + assertEquals(7L, ((Map) state).get("deserializationFailures")); + + // the key stays present once the count has been read + assertEquals(0L, ((Map) server.getState()).get("deserializationFailures")); + } finally { + server.close(); + } + } +}