This is an automated email from the ASF dual-hosted git repository.
rzo1 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/storm.git
The following commit(s) were added to refs/heads/master by this push:
new cccc23303 Drop malformed tuple payloads instead of killing the
receiving worker (#9076)
cccc23303 is described below
commit cccc23303d8d9ad3e2eaa530f1ee816ca336ff78
Author: L1nq <[email protected]>
AuthorDate: Tue Sep 8 21:47:48 2026 +0800
Drop malformed tuple payloads instead of killing the receiving worker
(#9076)
* 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
* 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
* Narrow the recv() try to deserialization so post-decode failures propagate
Tuple addressing and the metrics update run after the tuple has decoded; a
failure there is not a deserialization failure and must not drop a decoded
tuple or inflate the count.
---
docs/Metrics.md | 4 +-
.../messaging/DeserializingConnectionCallback.java | 88 +++++++-
.../org/apache/storm/messaging/netty/Server.java | 6 +
.../storm/serialization/KryoTupleDeserializer.java | 3 +
.../DeserializingConnectionCallbackTest.java | 234 +++++++++++++++++++++
.../apache/storm/messaging/netty/ServerTest.java | 43 ++++
6 files changed, 375 insertions(+), 3 deletions(-)
diff --git a/docs/Metrics.md b/docs/Metrics.md
index 8d620f0da..e7349f771 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 b038e026c..6a8464f43 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,38 @@ import org.apache.storm.task.GeneralTopologyContext;
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));
+
+ // Rate limits for drop logging; see recv().
+ 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 +74,12 @@ public class DeserializingConnectionCallback implements
IConnectionCallback, IMe
// 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,19 +90,63 @@ public class DeserializingConnectionCallback implements
IConnectionCallback, IMe
}
+ // 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());
+ Tuple tuple;
+ try {
+ tuple = des.deserialize(message.message());
+ } 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, which usually "
+ + "means a class is missing from the worker
classpath",
+ consecutiveDrops, e);
+ }
+ continue;
+ }
AddressedTuple addrTuple = new AddressedTuple(message.task(),
tuple);
updateMetrics(tuple.getSourceTask(), message);
ret.add(addrTuple);
+ if (consecutiveDropCount.get() != 0L) {
+ consecutiveDropCount.set(0L);
+ }
}
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.
*
@@ -91,6 +167,14 @@ public class DeserializingConnectionCallback implements
IConnectionCallback, IMe
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 da5adeacf..3d54cab0d 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 java.util.function.Supplier;
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 @@ class Server extends ConnectionWithStatus implements
IStatefulObject, ISaslServe
}
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 a310eac9c..301a8ca96 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 @@ public class KryoTupleDeserializer implements
ITupleDeserializer {
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);
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 f622ade12..c62043679 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,198 @@ public class DeserializingConnectionCallbackTest {
assertTrue(metrics instanceof Map);
assertEquals(6L, ((Map<?, ?>) metrics).get("123-456"));
}
+
+ @Test
+ public void testTruncatedKryoPayloadDroppedAndBatchContinues() {
+ Map<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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<String, Object> 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());
+ }
+
+ @Test
+ public void testPostDecodeFailurePropagatesAndIsNotCounted() {
+ WorkerState.ILocalTransferCallback transfer =
mock(WorkerState.ILocalTransferCallback.class);
+ // IllegalArgumentException is a tolerated deserialization-failure
type; throwing it from
+ // updateMetrics, which runs after a successful decode, proves the try
scope covers decoding only.
+ DeserializingConnectionCallback callback = new
DeserializingConnectionCallback(baseConf(), context, transfer) {
+ @Override
+ protected void updateMetrics(int sourceTaskId, TaskMessage
message) {
+ throw new IllegalArgumentException("injected after decode");
+ }
+ };
+
+ assertThrows(IllegalArgumentException.class,
+ () -> callback.recv(Collections.singletonList(
+ taskMessage(serializedTuple(baseConf(), new
Values("nathan", 1))))));
+
+ verify(transfer, never()).transfer(any());
+ assertEquals(0L, callback.getAndResetDeserializationFailures());
+ }
+
+ private void assertBatchDeliversOnlyValidMessages(Map<String, Object>
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<ArrayList<AddressedTuple>> captor =
ArgumentCaptor.forClass(ArrayList.class);
+ verify(transfer).transfer(captor.capture());
+ List<AddressedTuple> 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<String, Object> baseConf() {
+ Map<String, Object> conf = new HashMap<>(Utils.readStormConfig());
+ return conf;
+ }
+
+ private byte[] serializedTuple(Map<String, Object> conf, List<Object>
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 000000000..f1f7ca6fe
--- /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();
+ }
+ }
+}
