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 118240016 Throw a typed exception for tuples with unknown task or
stream ids and add a deserialization strict mode (#9094)
118240016 is described below
commit 118240016a3f555db35710a5b5bb41b834673ef6
Author: L1nq <[email protected]>
AuthorDate: Mon Oct 12 01:36:37 2026 +0800
Throw a typed exception for tuples with unknown task or stream ids and add
a deserialization strict mode (#9094)
* Throw a typed exception for tuples with unknown task or stream ids and
add a deserialization strict mode
* Make TupleDeserializationException extend IllegalArgumentException and
document the strict mode
The exception thrown for unknown task or stream ids now extends
IllegalArgumentException, so handlers written against 3.1.0 keep matching,
and it gains a (String, Throwable) constructor. The strict mode flag is
documented in Serialization.md and Metrics.md, and the config javadoc spells
out that a persistent bad frame puts the worker in a restart loop. The
unknown-stream test no longer pins the exact message text and the strict
mode is now covered for the typed exception as well.
* Return null for unknown components in IdDictionary and document the
strict-mode boundary for IOException-wrapped failures
The tolerated-failures set matches TupleDeserializationException through its
IllegalArgumentException supertype, so the explicit entry and the unused
two-argument constructor are removed. IdDictionary.getStreamName returns
null
when the component is unknown, so both unresolved-routing cases reach the
typed
exception instead of a NullPointerException that stays fatal in default
mode.
The strict-mode documentation now separates the two outcomes: decode
failures
exit the worker, while failures wrapping an IOException close the
connection and
drop the batch with the worker kept running. The unknown-stream test
derives its
id from the declared stream count instead of a hardcoded constant.
---
conf/defaults.yaml | 1 +
docs/Metrics.md | 2 +-
docs/Serialization.md | 1 +
storm-client/src/jvm/org/apache/storm/Config.java | 13 +++++
.../messaging/DeserializingConnectionCallback.java | 8 ++-
.../storm/serialization/KryoTupleDeserializer.java | 5 +-
.../storm/serialization/SerializationFactory.java | 3 +-
.../TupleDeserializationException.java | 22 ++++++++
.../DeserializingConnectionCallbackTest.java | 66 ++++++++++++++++++++--
.../serialization/SerializationFactoryTest.java | 12 ++++
10 files changed, 123 insertions(+), 10 deletions(-)
diff --git a/conf/defaults.yaml b/conf/defaults.yaml
index 6fd7a04b9..939d79b1e 100644
--- a/conf/defaults.yaml
+++ b/conf/defaults.yaml
@@ -61,6 +61,7 @@ storm.compression.zstd.level: 3
storm.compression.zstd.max.decompressed.bytes: 104857600
storm.compression.gzip.max.decompressed.bytes: 104857600
topology.tuple.compression.max.decompressed.bytes: 10485760
+topology.tuple.deserialization.strict.enable: false
storm.codedistributor.class:
"org.apache.storm.codedistributor.LocalFileSystemCodeDistributor"
storm.workers.artifacts.dir: "workers-artifacts"
storm.health.check.dir: "healthchecks"
diff --git a/docs/Metrics.md b/docs/Metrics.md
index e7349f771..9a3a9fc83 100644
--- a/docs/Metrics.md
+++ b/docs/Metrics.md
@@ -302,7 +302,7 @@ Be aware that the `__system` bolt is an actual bolt so
regular bolt metrics desc
`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.
+`deserializationFailures` is the number of incoming messages that failed to
deserialize and were dropped. When
`topology.tuple.deserialization.strict.enable` is set, deserialization failures
are not dropped or counted; they propagate instead. Non-IO failures terminate
the worker, while failures caused by an `IOException`, such as a corrupt
compressed frame, close the connection and leave the worker running.
##### Send (Netty Client)
diff --git a/docs/Serialization.md b/docs/Serialization.md
index b8701e8aa..4d3841ad8 100644
--- a/docs/Serialization.md
+++ b/docs/Serialization.md
@@ -131,6 +131,7 @@ Be aware that the topology-wide form enables compression
for *every* remote-boun
| `topology.tuple.compression.threshold` | `1460` | Minimum serialized tuple
size, in bytes, before compression is attempted. Tuples at or below this size
are sent uncompressed. The default matches the typical Ethernet TCP MSS, so
payloads that already fit in a single network frame are never compressed. |
| `storm.compression.zstd.level` | `3` | Zstd compression level. Supported
range is 1–19; levels 20–22 (ultra mode) are prohibited because of their memory
requirements. |
| `topology.tuple.compression.max.decompressed.bytes` | `10485760` (10 MB) |
Upper bound on the decompressed size of a single tuple. Decompression that
would exceed this limit fails, guarding against malicious or corrupt payloads. |
+| `topology.tuple.deserialization.strict.enable` | `false` | Makes tuple
deserialization failures fatal: an incoming message that fails to decode kills
the receiving worker instead of being dropped and counted. This restores the
pre-3.1.0 behavior and is mainly useful for debugging, because a persistent bad
frame will put the worker in a restart loop. A failure rooted in an
`IOException`, such as a corrupt or oversized compressed frame, is an
exception: the receiving worker closes that c [...]
#### How decompression works
diff --git a/storm-client/src/jvm/org/apache/storm/Config.java
b/storm-client/src/jvm/org/apache/storm/Config.java
index 519b8d943..9c275f82c 100644
--- a/storm-client/src/jvm/org/apache/storm/Config.java
+++ b/storm-client/src/jvm/org/apache/storm/Config.java
@@ -1648,6 +1648,19 @@ public class Config extends HashMap<String, Object> {
*/
@IsPositiveNumber(includeZero = false)
public static final String
TOPOLOGY_TUPLE_COMPRESSION_MAX_DECOMPRESSED_BYTES =
"topology.tuple.compression.max.decompressed.bytes";
+ /**
+ * Topology configuration to make tuple deserialization failures fatal
instead of dropping the undecodable message.
+ * By default a message that fails to decode on the receiving worker is
dropped and counted, and the worker keeps
+ * running. When set to {@code true}, any deserialization failure
propagates and the worker exits, restoring the
+ * pre-3.1.0 behavior. Be aware that a single corrupt frame from a peer
then kills the worker, and the supervisor
+ * restarts it into the same failure, so a persistent bad frame results in
a restart loop. Failures that wrap an
+ * {@code IOException}, such as a broken or oversized compressed frame, do
not reach the exit path: the messaging
+ * layer treats them like a connection error, drops the batch being read,
and leaves the worker running while the
+ * peer reconnects.
+ * Default: {@code false}.
+ */
+ @IsBoolean
+ public static final String TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE =
"topology.tuple.deserialization.strict.enable";
/**
* Configure the topology metrics reporters to be used on workers.
*/
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 6a8464f43..cee8776a1 100644
---
a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java
+++
b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java
@@ -43,7 +43,8 @@ public class DeserializingConnectionCallback implements
IConnectionCallback, IMe
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.
+ // the fatal handling in StormServerHandler. TupleDeserializationException
is thrown by KryoTupleDeserializer
+ // for unknown task or stream ids.
private static final Set<Class<?>> TOLERATED_DESERIALIZATION_FAILURES =
new HashSet<>(Arrays.asList(
IOException.class,
KryoException.class,
@@ -71,6 +72,8 @@ public class DeserializingConnectionCallback implements
IConnectionCallback, IMe
}
};
+ private final boolean strictMode;
+
// Track serialized size of messages.
private final boolean sizeMetricsEnabled;
private final ConcurrentHashMap<String, AtomicLong> byteCounts = new
ConcurrentHashMap<>();
@@ -87,6 +90,7 @@ public class DeserializingConnectionCallback implements
IConnectionCallback, IMe
this.context = context;
cb = callback;
sizeMetricsEnabled =
ObjectReader.getBoolean(conf.get(Config.TOPOLOGY_SERIALIZED_MESSAGE_SIZE_METRICS),
false);
+ strictMode =
ObjectReader.getBoolean(conf.get(Config.TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE),
false);
}
@@ -104,7 +108,7 @@ public class DeserializingConnectionCallback implements
IConnectionCallback, IMe
try {
tuple = des.deserialize(message.message());
} catch (Exception e) {
- if (!isToleratedDeserializationFailure(e)) {
+ if (strictMode || !isToleratedDeserializationFailure(e)) {
throw e;
}
deserializationFailures.incrementAndGet();
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 301a8ca96..e433603f1 100644
---
a/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
+++
b/storm-client/src/jvm/org/apache/storm/serialization/KryoTupleDeserializer.java
@@ -79,9 +79,12 @@ public class KryoTupleDeserializer implements
ITupleDeserializer {
int streamId = kryoInput.readInt(true);
String componentName = context.getComponentId(taskId);
if (componentName == null) {
- throw new IllegalArgumentException("Received a tuple from
unknown task " + taskId);
+ throw new TupleDeserializationException("Received a tuple from
unknown task " + taskId);
}
String streamName = ids.getStreamName(componentName, streamId);
+ if (streamName == null) {
+ throw new TupleDeserializationException("Component " +
componentName + " has no stream with id " + streamId);
+ }
MessageId id = MessageId.deserialize(kryoInput);
List<Object> values = kryo.deserializeFrom(kryoInput);
return new TupleImpl(context, values, componentName, taskId,
streamName, id);
diff --git
a/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java
b/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java
index 18b471c99..9180978ea 100644
---
a/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java
+++
b/storm-client/src/jvm/org/apache/storm/serialization/SerializationFactory.java
@@ -259,7 +259,8 @@ public class SerializationFactory {
}
public String getStreamName(String component, int stream) {
- return streamIdToName.get(component).get(stream);
+ Map<Integer, String> streamNames = streamIdToName.get(component);
+ return streamNames == null ? null : streamNames.get(stream);
}
}
}
diff --git
a/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java
b/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java
new file mode 100644
index 000000000..8329ea198
--- /dev/null
+++
b/storm-client/src/jvm/org/apache/storm/serialization/TupleDeserializationException.java
@@ -0,0 +1,22 @@
+/**
+ * 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
distributed under the License 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.serialization;
+
+/**
+ * Thrown when a serialized tuple names a source task or stream that the
receiving topology cannot resolve.
+ */
+public class TupleDeserializationException extends IllegalArgumentException {
+ public TupleDeserializationException(String message) {
+ super(message);
+ }
+}
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 2a982b3a3..de8bf62b2 100644
---
a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java
+++
b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java
@@ -27,6 +27,7 @@ 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.serialization.TupleDeserializationException;
import org.apache.storm.task.GeneralTopologyContext;
import org.apache.storm.testing.TestWordCounter;
import org.apache.storm.testing.TestWordSpout;
@@ -132,14 +133,62 @@ public class DeserializingConnectionCallbackTest {
@Test
public void testUnknownSourceTaskDroppedAndBatchContinues() {
Map<String, Object> conf = baseConf();
+
+ TupleDeserializationException thrown =
assertThrows(TupleDeserializationException.class,
+ () -> new
KryoTupleDeserializer(conf, context).deserialize(unknownSourceTaskTuple()));
+ assertTrue(thrown.getMessage().contains("9999"),
+ "expected the task id in the message but was: " +
thrown.getMessage());
+
+ assertBatchDeliversOnlyValidMessages(conf, unknownSourceTaskTuple());
+ }
+
+ @Test
+ public void testUnknownStreamIdDroppedAndBatchContinues() {
+ Map<String, Object> conf = baseConf();
+ // IdDictionary assigns ids 1..n to the component's n declared
streams, so the first id past that count is undeclared.
+ int undeclaredStreamId =
Utils.getComponentCommon(context.getRawTopology(),
SOURCE_COMPONENT).get_streams().size() + 1;
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();
+ out.writeInt(SOURCE_TASK_ID, true); // source task that exists in the
topology
+ out.writeInt(undeclaredStreamId, true);
+ byte[] unknownStream = out.toBytes();
- assertThrows(IllegalArgumentException.class, () -> new
KryoTupleDeserializer(conf, context).deserialize(unknownTask));
+ TupleDeserializationException thrown =
assertThrows(TupleDeserializationException.class,
+ () -> new
KryoTupleDeserializer(conf, context).deserialize(unknownStream));
+ assertTrue(thrown.getMessage().contains(SOURCE_COMPONENT),
+ "expected the component name in the message but was: " +
thrown.getMessage());
- assertBatchDeliversOnlyValidMessages(conf, unknownTask);
+ assertBatchDeliversOnlyValidMessages(conf, unknownStream);
+ }
+
+ @Test
+ public void testStrictModeMakesFailuresFatal() {
+ Map<String, Object> conf = baseConf();
+ conf.put(Config.TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE, true);
+ byte[] full = serializedTuple(conf, new
Values("a-string-long-enough-to-survive-truncation", 7));
+ byte[] truncated = Arrays.copyOf(full, full.length - 10);
+
+ WorkerState.ILocalTransferCallback transfer =
mock(WorkerState.ILocalTransferCallback.class);
+ DeserializingConnectionCallback callback = new
DeserializingConnectionCallback(conf, context, transfer);
+
+ assertThrows(KryoException.class, () ->
callback.recv(Collections.singletonList(taskMessage(truncated))));
+
+ verify(transfer, never()).transfer(any());
+ assertEquals(0L, callback.getAndResetDeserializationFailures());
+ }
+
+ @Test
+ public void testStrictModeMakesUnknownTaskFailureFatal() {
+ Map<String, Object> conf = baseConf();
+ conf.put(Config.TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE, true);
+
+ WorkerState.ILocalTransferCallback transfer =
mock(WorkerState.ILocalTransferCallback.class);
+ DeserializingConnectionCallback callback = new
DeserializingConnectionCallback(conf, context, transfer);
+
+ assertThrows(TupleDeserializationException.class,
+ () ->
callback.recv(Collections.singletonList(taskMessage(unknownSourceTaskTuple()))));
+
+ verify(transfer, never()).transfer(any());
+ assertEquals(0L, callback.getAndResetDeserializationFailures());
}
@Test
@@ -269,6 +318,13 @@ public class DeserializingConnectionCallbackTest {
assertNull(callback.getValueAndReset());
}
+ private static byte[] unknownSourceTaskTuple() {
+ 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
+ return out.toBytes();
+ }
+
private Map<String, Object> baseConf() {
Map<String, Object> conf = new HashMap<>(Utils.readStormConfig());
return conf;
diff --git
a/storm-client/test/jvm/org/apache/storm/serialization/SerializationFactoryTest.java
b/storm-client/test/jvm/org/apache/storm/serialization/SerializationFactoryTest.java
index d7e7bfcd4..998098643 100644
---
a/storm-client/test/jvm/org/apache/storm/serialization/SerializationFactoryTest.java
+++
b/storm-client/test/jvm/org/apache/storm/serialization/SerializationFactoryTest.java
@@ -16,11 +16,14 @@ import com.esotericsoftware.kryo.Kryo;
import java.util.Map;
import org.apache.storm.Config;
import org.apache.storm.security.serialization.BlowfishTupleSerializer;
+import org.apache.storm.testing.TestWordSpout;
+import org.apache.storm.topology.TopologyBuilder;
import org.apache.storm.utils.ListDelegate;
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.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
public class SerializationFactoryTest {
@@ -65,6 +68,15 @@ public class SerializationFactoryTest {
kryo.getSerializer(MockSerObject.class).getClass());
}
+ @Test
+ public void test_get_stream_name_null_for_unknown_component() {
+ TopologyBuilder builder = new TopologyBuilder();
+ builder.setSpout("spout", new TestWordSpout(true), 1);
+ SerializationFactory.IdDictionary ids = new
SerializationFactory.IdDictionary(builder.createTopology());
+
+ assertNull(ids.getStreamName("no-such-component", 1));
+ }
+
public static class MockKryoDecorator implements IKryoDecorator {
@Override
public void decorate(Kryo k, Map<String, Object> conf) {
