Throw a typed exception for tuples with unknown task or stream ids and add a deserialization strict mode

[L1nq0]

Relates to #9077

This implements part of #9077: the unknown-task and unknown-stream checks in KryoTupleDeserializer now throw a typed exception, and a new config makes deserialization failures fatal again.

What changed

KryoTupleDeserializer throws TupleDeserializationException (a RuntimeException in org.apache.storm.serialization) when a tuple names a source task the receiving topology cannot resolve, or a stream id the source component does not declare. The unknown-task case previously threw a bare IllegalArgumentException; the unknown-stream case previously resolved to a null stream name and the tuple was delivered anyway.

DeserializingConnectionCallback treats the new exception as a tolerated deserialization failure, so both cases are dropped and counted like the other decode failures, with the task or stream id in the message.

The new config topology.tuple.deserialization.strict.enable (default false) makes any deserialization failure propagate instead of being dropped, restoring the pre-3.1.0 fail-fast behavior. The default keeps the behavior introduced by #9076.

Behavior changes

• The exception type for an unresolvable source task changes from IllegalArgumentException to TupleDeserializationException. The message text is unchanged.
• A tuple whose stream id does not resolve for its source component is now dropped with an error naming the component and the stream id. Previously it was delivered with a null stream name.

Tests

DeserializingConnectionCallbackTest gains a test for the unknown-stream drop and one for strict mode making a truncated payload fatal. The existing unknown-task test now asserts the typed exception and that the message names the task id. The storm-client test suite passes.

Open questions

Whether the other tolerated exception types in DeserializingConnectionCallback should also be replaced by typed exceptions is left open; this change only types the failures KryoTupleDeserializer itself detects. The config flag name follows the existing topology.*.enable convention; happy to rename if maintainers prefer another form.

[rzo1]

The unknown-stream check is correct and closes a real hole: a null stream name was previously passed to TupleImpl and the tuple delivered anyway. Sender and receiver build IdDictionary from the same topology and KryoTupleSerializer would fail on the sending side before emitting an unresolvable id, so a null on the receiving side does mean a corrupt or mismatched frame. No concerns there.

Three things before this can go in.

1. TupleDeserializationException extends RuntimeException. The unknown-task case threw IllegalArgumentException in 3.1.0, so anything catching that stops matching. Please extend IllegalArgumentException instead. Existing callers keep working, and the entry you added to TOLERATED_DESERIALIZATION_FAILURES becomes redundant, though keeping it explicit is fine. While you are in there, add a (String, Throwable) constructor.

2. The config is undocumented. topology.tuple.deserialization.strict.enable exists only in Config.java and defaults.yaml. docs/Serialization.md carries the table where topology.tuple.compression.max.decompressed.bytes is documented, and docs/Metrics.md covers deserializationFailures, which you added in Drop malformed tuple payloads instead of killing the receiving worker #9076. The flag belongs in both.

3. The config javadoc does not say what enabling it costs. Under strict mode a single corrupt frame from a peer kills the worker, the supervisor restarts it, and the same frame kills it again. That restart loop is what Drop malformed tuple payloads instead of killing the receiving worker #9076 fixed, and users should read it in the config description rather than infer it. The "pre-3.1.0 behavior" wording is accurate, Drop malformed tuple payloads instead of killing the receiving worker #9076 is contained in v3.1.0, so keep that and add the consequence.

Minor:

• assertTrue(thrown.getMessage().contains("id 3")) ties the test to the exact message text. The exception type plus the component name would be enough.
• No test covers strict mode with the new exception. testStrictModeMakesFailuresFatal only exercises the truncated payload path.

On your open question: topology.tuple.deserialization.strict.enable is fine, it matches the existing keys.

[L1nq0]

@rzo1 Thanks for the review. Commit be82785 addresses everything.

TupleDeserializationException now extends IllegalArgumentException and has a (String, Throwable) constructor, so handlers written against the 3.1.0 behavior keep matching. The entry in TOLERATED_DESERIALIZATION_FAILURES stays; it is redundant now but it documents where the exception comes from.

The config is documented in the table in docs/Serialization.md and next to deserializationFailures in docs/Metrics.md. The javadoc keeps the pre-3.1.0 wording and now states the consequence: a single corrupt frame kills the worker and the supervisor restarts it into the same failure, so a persistent bad frame results in a restart loop.

On the test points: the unknown-stream case no longer pins the message text, it checks the exception type and the component name, and testStrictModeMakesUnknownTaskFailureFatal covers the strict mode with the typed exception thrown for an unknown task id.

The config name stays topology.tuple.deserialization.strict.enable, per your confirmation.