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 b038e026c2b..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,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> 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 conf; private final GeneralTopologyContext context; - private final ThreadLocal des = + private ThreadLocal des = new ThreadLocal() { @Override protected KryoTupleDeserializer initialValue() { @@ -47,7 +77,12 @@ 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); + // 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) { @@ -58,23 +93,65 @@ 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(); 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); + 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. * 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 f622ade12ce..b6c36643ced 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,70 @@ 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.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 +117,188 @@ 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(IllegalArgumentException.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 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")); + 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 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()); + + assertEquals(1L, callback.getAndResetDeserializationFailures()); + 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 { + } } 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(); + } + } +}