Drop malformed tuple payloads instead of killing the receiving worker

[L1nq0]

Closes #9074

Upgrade note: a malformed message body no longer kills the receiving worker, including the IOException case. Previously a channel-level IOException closed the connection and lost every message still queued on it; now the one undecodable message is dropped and the connection, along with the rest of the batch, keeps going. Payloads that used to tear down a connection mid-stream will instead show up as deserializationFailures counts, rate-limited ERROR logs, and a WARN when failures persist.

What this changes

DeserializingConnectionCallback.recv() wraps only the deserialization of each message in a try/catch. When the failure matches a known decode-time exception type, the message is dropped and the loop continues with the next message in the batch. Tuple addressing and metrics updates run after the try, so a failure there propagates rather than dropping a validly decoded tuple or inflating the count. Anything outside the tolerated set is rethrown and keeps today's behavior: StormServerHandler.exceptionCaught, then Utils.handleUncaughtException, then worker exit. Errors are never caught.

The tolerated set is the set of exception types a garbage or hostile byte stream can already produce during tuple decode, matched by walking each failure's cause chain:

IOException, KryoException, IllegalArgumentException, NegativeArraySizeException, ClassCastException, ArrayIndexOutOfBoundsException, BufferUnderflowException, ClassNotFoundException

Mapping to wire-level failure modes: truncated or negative length fields (ArrayIndexOutOfBoundsException, NegativeArraySizeException, BufferUnderflowException), unregistered classes under topology.kryo.register with registration required (IllegalArgumentException, thrown directly by Kryo.getRegistration when a name reference resolves to a class with no registration), classes present only on the sending side (ClassNotFoundException), wrong runtime types (ClassCastException), and kryo's own decode failures (KryoException) plus underlying stream problems (IOException). A tuple whose source task id does not exist in the topology is rejected up front by KryoTupleDeserializer with an IllegalArgumentException naming the task, instead of coming out of the stream lookup as a NullPointerException.

I did not broaden StormServerHandler.ALLOWED_EXCEPTIONS. That whitelist is about transport-level channel errors and closing the connection; a payload that fails to decode is an application-level event, and handling it where the decode happens is what keeps the rest of the batch alive. The whitelist stays as the backstop for anything that still escapes.

Observability

Drop logging is rate-limited. The first 10 failures are logged in full; after that, one summary line per 100 further failures carries the running total. A run of 1000 consecutive failures without a successful deserialization additionally logs a WARN, and the consecutive count resets on the next success. This keeps a hostile peer from filling the log volume, and keeps a real problem such as a class missing from the worker classpath visible as a WARN rather than silent no-progress with a stack trace per tuple. Failures are also counted in a deserializationFailures counter published as a top-level key of __recv-iconnection through Server.getState(), always present including 0, and documented in docs/Metrics.md.

What this does not fix

A frame declaring a huge positive array or collection length can still kill the worker: kryo pre-allocates from the length field, the allocation fails with an OutOfMemoryError, and recv() catches Exception, not Error. Catching Error would be wrong. NegativeArraySizeException, the wrap-around case, is tolerated; the large-positive case is not, and this is unchanged from master. Bounding allocations on the kryo path, along the lines of the maxarray and maxbytes limits the java-serialization fallback bridge gets in #9075, is follow-up work.

Interaction with at-least-once spouts

A message that deterministically fails to decode is dropped but never acked. With the kafka spout in at-least-once mode the offset never commits, uncommitted offsets on that partition climb to maxUncommittedOffsets (default 10,000,000), and the spout stops polling the partition, so the topology makes no progress on it and the symptom is climbing lag. On master the same input produces a worker crash loop that also never commits the offset. The real fix is an eventual permanent drop or dead-letter so the offset can advance; that is follow-up work as well.

Tests

8 cases in DeserializingConnectionCallbackTest, driving recv() through the real deserializer where possible: one tolerant drop per interesting exception type (verify log, counter, and batch continuation), an IOException mid-batch that skips exactly the bad message while a good message later in the same batch still lands, rethrow of a non-tolerated exception, a failure thrown after a successful decode propagating without being dropped or counted as a deserialization failure, and the failure counter staying separate from the size metrics and resetting after a read.

Also verified on a live 3.0.1-SNAPSHOT cluster: replaying a 27-byte frame carrying an unregistered class against a worker port kills the worker on master (terminating server, then process exit and supervisor restart) and on this branch produces the log line "Failed to deserialize a message of 27 bytes destined for task 2, dropping it" with the worker surviving and the topology staying active.

[rzo1]

Thanks — this is a well-scoped change and the reasoning in the description is sound. I built the branch and ran the new tests locally on JDK 25: 9/9 pass. Note that CI has not run on this PR yet (fork workflows need approval here); I'll get that triggered.

Two things I want to endorse explicitly, because they are the parts someone will inevitably want to change:

Not widening StormServerHandler.ALLOWED_EXCEPTIONS is correct. That whitelist is about transport-level channel errors, and handling a decode failure where the decode happens is what keeps the rest of the batch alive. MessageDecoder-level failures already land in MessageDecoder.exceptionCaught → ctx.close(), so recv() really is the only path that kills the worker. Please keep it as is.

Continuing the loop after a mid-stream failure is safe, which was my main correctness worry — the thread-local KryoTupleDeserializer is reused for the next message in the batch. I checked: Kryo 5.6.2 resets in a finally (Kryo.java:629), so the class resolver isn't left dirty. Your testJavaFallbackMissingClassDroppedAndBatchContinues case exercises exactly the name-based path where that would have bitten. Good test choice.

Now the review proper.

1. The remaining DoS is log volume

"One frame kills the worker" becomes "one ERROR with a full stack trace per frame". An attacker who can reach the worker port now fills the log volume instead of killing the process — better, but still an unauthenticated remote knob on a shared disk.

Please rate-limit: log the first N and then periodically with a running total. There's precedent in WorkerState.dropMessage (WorkerState.java:622), which carries a running dropCount in the message.

The same applies to the non-hostile case, which I think matters more day to day: a class genuinely missing from the worker classpath now produces ClassNotFoundException on every tuple. The topology silently makes no progress while emitting one stack trace per tuple, where before it failed loudly and obviously. A WARN on N consecutive failures would keep that visible in an operator's face.

2. NullPointerException in the tolerated set is papering over a missing check

The failure mode you map it to — a bogus task id inside a tuple — comes from KryoTupleDeserializer.deserializeTuple (KryoTupleDeserializer.java:79-80): context.getComponentId(taskId) returns null for an unknown id, and then ids.getStreamName(null, streamId) NPEs. That's two lines of validation at the source throwing a typed exception, rather than blanket-tolerating NPE at the callback.

It matters because Utils.exceptionCauseIsInstanceOf walks the entire cause chain, and user-supplied serializers registered via topology.kryo.register run inside des.deserialize. A real bug in one of those that surfaces as an NPE (or IllegalArgumentException) is now silently dropped data rather than a loud failure — the opposite of what you want for a topology-logic bug.

The durable shape is a dedicated TupleDeserializationException wrapping everything the decode path can legitimately throw, caught narrowly here. If that's more than you want to take on in this PR, dropping NullPointerException from the set and adding the explicit task-id check would get most of the way.

3. The metric lands in the wrong namespace

deserializationFailures is merged into the same map as the "srcTask-destTask" byte counts, and Server.getState() publishes that whole map under "messageBytes" (Server.java:245-249). So it surfaces as __recv-iconnection.messageBytes.deserializationFailures, inside a namespace whose keys are otherwise src-dest pairs. Nothing in-tree parses those keys, but external metrics consumers do.

Put it at the top level of Server.getState() instead (ret.put("deserializationFailures", ...)) and leave getValueAndReset() returning null when size metrics are off — that keeps the existing contract untouched.

Related: emitting the key only when failures > 0 makes it appear and disappear between reporting buckets, and most backends want a stable 0. Moving it out of messageBytes lets you always emit it and solves both at once.

Also __recv-iconnection is documented at docs/Metrics.md:291; a new user-visible counter should get a line there.

Smaller things

• Import order: com.esotericsoftware.kryo.KryoException sits after the java.* block in DeserializingConnectionCallback.java but correctly before it in the test. CustomImportOrder is severity=warning so it won't fail the build, just inconsistent within the one PR.
• The test reflectively writes the private final ThreadLocal des field. It works on JDK 25 — I ran it — but final-field reflection is exactly what JEP 500's integrity-by-default is aimed at, and this will become a maintenance problem. A package-private constructor taking the deserializer would outlive it.
• isToleratedDeserializationFailure walks the full cause chain once per entry in the set, so nine passes per failure. Irrelevant at current volumes, and if you rate-limit the logging it stays irrelevant — mentioning it only because a single pass over the chain testing instanceof against the list would be simpler to read.

Merge order

I'd like to land this before #9075. That PR's filter rejects a payload with an InvalidClassException, which on master propagates up through recv() and kills the worker — turning an RCE into a remote worker-kill. InvalidClassException extends IOException, so with this change in first, a filtered payload is dropped and counted instead. The two compose nicely in that order.

[L1nq0]

Thanks for the careful review, and for independently checking the two spots I was most concerned about: the Kryo reset on a mid-stream failure and the MessageDecoder path. Your read matches what I found, so it's good to have it confirmed.

On the numbered points:

1. Log volume: agreed, that's the remaining DoS and it cuts both ways. Plan: the first few dropped messages get individual ERROR logs, then a periodic summary line carrying the running total, following the dropCount precedent in WorkerState.dropMessage. On top of that a WARN after N consecutive failures, so the genuine missing-class case surfaces as a loud signal instead of silent no-progress with a stack trace per tuple.

2. NullPointerException: agreed that tolerating it at the callback papers over a missing check, and with the broader point about user-supplied serializers: a real bug in one of those surfacing as an NPE or IAE should fail loudly, not vanish into dropped data. Taking the minimal path in this PR: NPE comes out of the tolerated set, and the task-id lookup in KryoTupleDeserializer gets an explicit check that throws a typed exception with the id in the message, which then flows through the tolerated path as a normal decode failure. The dedicated TupleDeserializationException wrapping everything the decode path can legitimately throw is the more durable shape for the broader concern; I'd rather track that as a follow-up issue than widen this PR's blast radius mid-review, but I'm happy either way.

3. Metric namespace: agreed on all three counts. The counter moves to the top level of Server.getState, it is always emitted with a stable zero rather than appearing only on failure, and docs/Metrics.md gets the line.

Smaller things, all accepted: import order fixed to match the test file, the reflective write to the ThreadLocal replaced by a package-private constructor taking the deserializer, and the tolerance check rewritten as a single pass over the cause chain testing instanceof against the list.

Merge order: agreed. This PR stays independent of #9075, and I'll rebase #9075 onto it after it lands. Thanks also for triggering the workflows.

[L1nq0]

The revisions are pushed.

Drop logging is now rate-limited: the first 10 failures log individually, then one
summary per 100 further failures carries the running total, and a WARN fires after
1000 consecutive failures; any successful deserialization resets that counter.

NullPointerException is out of the tolerated set, and KryoTupleDeserializer rejects
an unknown source task with an IllegalArgumentException naming the task.

deserializationFailures is a top-level key of __recv-iconnection now, always present
including 0, with getValueAndReset() back to the original contract. Metrics.md
documents the key.

The full storm-client suite is green locally (680 tests) with no new checkstyle
findings.

[reiabreu]

Note: drafted with the help of an LLM.

First of all, thank you for this submission. It's a great addition.

Mostly additions to rzo1's review:

1. The DoS is reduced, not closed — a large positive array length still OOMs the worker. recv() catches Exception, correctly not Error, but a malformed frame can declare a huge positive array/collection length that Kryo pre-allocates → OutOfMemoryError → worker dies. NegativeArraySizeException (the wrap-around case) is caught; the large-positive case isn't, and catching Error would be wrong. The fix is bounding allocations (Kryo input limits; the maxarray limit on Harden the java serialization fallback bridge with a JEP-290 serial filter #9075's fallback path). Worth naming as a known residual.

2. Interaction with contiguous-commit spouts (Kafka): drop-and-continue can become a silent stall. A deterministically undecodable message (e.g. a class missing on the receiving worker's classpath) is dropped but never acked, so its offset never commits. In at-least-once mode uncommitted offsets on that partition climb to maxUncommittedOffsets (default 10,000,000, KafkaSpoutConfig.java:47), and getPollablePartitionsInfo() (KafkaSpout.java:325) then stops polling it — churning on the poison offset, ingesting nothing new. So we trade a worker crash for a quiet per-partition stall (visible mainly as climbing lag). The durable fix is an eventual permanent-drop / dead-letter so the offset can advance — reasonable as a follow-up, but worth stating.

3. Following up on @rzo1's point 2 — the generic types still carry the swallow-a-real-bug risk. Dropping NullPointerException and adding the task-id check handled that path, but IllegalArgumentException, ClassCastException, and ArrayIndexOutOfBoundsException stay in the set, and a user serializer (via topology.kryo.register, run inside des.deserialize) that throws one on a real bug is still silently dropped and counted as a deserialization failure. The TupleDeserializationException shape you suggested is the clean fix. Sketch:

   public class TupleDeserializationException extends RuntimeException {
       public TupleDeserializationException(String message) { super(message); }
       public TupleDeserializationException(String message, Throwable cause) { super(message, cause); }
   }

   In KryoTupleDeserializer.deserializeTuple — validate structural fields explicitly (so no reliance on a generic exception) and wrap only the corruption-indicating types:

   try {
       kryoInput.setBuffer(data, 0, data.length);
       int taskId = kryoInput.readInt(true);
       int streamId = kryoInput.readInt(true);
       String componentName = context.getComponentId(taskId);
       if (componentName == null) {
           throw new TupleDeserializationException("Tuple from unknown task " + taskId);
       }
       String streamName = ids.getStreamName(componentName, streamId);
       if (streamName == null) {
           throw new TupleDeserializationException(
               "Unknown stream id " + streamId + " for component " + componentName);
       }
       MessageId id = MessageId.deserialize(kryoInput);
       List<Object> values = kryo.deserializeFrom(kryoInput);
       return new TupleImpl(context, values, componentName, taskId, streamName, id);
   } catch (KryoException | IOException | BufferUnderflowException | NegativeArraySizeException e) {
       throw new TupleDeserializationException("Failed to deserialize tuple", e);
   }

   Then recv() catches exactly one type — no cause-chain walk, no generic-type set:

   } catch (TupleDeserializationException e) {
       // drop + count + log
   }

   One honest caveat: kryo.deserializeFrom runs both the framework decode and user serializers with no seam between them, so a generic exception thrown there is inherently ambiguous. The gain is that (a) structural failures are now typed at the source, letting you delete NullPointerException outright rather than tolerate it, and (b) recv() catches one intentional type instead of pattern-matching generics — so a user-serializer bug throwing a raw ArrayIndexOutOfBoundsException propagates loudly instead of being swallowed. I'd deliberately not wrap the ambiguous generics (IllegalArgumentException/ClassCastException/ArrayIndexOutOfBoundsException); worst case an unregistered-class IllegalArgumentException from Kryo surfaces loudly, which is a fair trade against silently dropping real bugs.

4. Minor — try scope. updateMetrics(...) and ret.add(...) are inside the try, so a post-decode failure there would drop a validly decoded tuple and miscount it. Narrow the try to des.deserialize(...).

5. Minor — stale description. The PR body still lists NullPointerException (9 types); the code has 8. Update to match.

Thanks for the careful work.

[L1nq0]

@reiabreu Thanks for this review, and especially for working through the two interaction cases I had not pushed to the end, the allocation path and the at-least-once spout behavior. Both are now stated in the description as limitations rather than left implicit.

The large-positive-length OOM: agreed, and agreed that catching Error is not an answer. The description now carries it as a limitation, unchanged from master. The bounding work, kryo input-side limits and the maxarray limit the fallback bridge gets in #9075, is where the actual fix belongs; I have it on the follow-up list rather than in this PR.

The spout interaction: agreed with the mechanism as you describe it, and the description now states it. One addition worth making explicit: on master the same deterministically undecodable message does not stall the partition quietly. It kills the worker, the supervisor restarts it, the spout re-emits after the tuple times out, and it kills it again, a crash loop that also never commits the offset. So this PR trades a loud crash loop for a quiet stall, which is the same trade it makes everywhere else, and the real fix, an eventual permanent drop or dead-letter so the offset can advance, is the second follow-up I want to file.

The generic-type concern and the TupleDeserializationException sketch: I agree with the concern, and I want to push back on one boundary in the sketch.

In kryo 5.6.2, when registration is required and a name reference resolves to a class with no registration, the IllegalArgumentException ("Class is not registered: ...") is thrown directly by Kryo.getRegistration(Class), not wrapped in a KryoException. DefaultClassResolver wraps only the unknown-class-id and name-lookup failures. That raw IllegalArgumentException is exactly what the 27-byte frame in the description produces on the live cluster (Storm 3.x requires registration unless the java serialization fallback is enabled), and it is the vector #9074 was filed on. With the sketch leaving IllegalArgumentException unwrapped by design, that frame goes back to killing the worker, which reopens the remote worker-kill this PR closes and also undercuts the compose-with-#9075 ordering rzo1 described, since a payload the serial filter rejects with an InvalidClassException needs the drop path to survive.

So the trade I would rather take: keep IllegalArgumentException tolerated in this PR, accept that a user serializer bug that raises one of the generic types is dropped and counted rather than loud, and keep the signal that is already built in, the WARN at 1000 consecutive failures and the always-present counter. Then do the exception-type redesign properly in a follow-up issue where the wrap boundary is the actual design question. Your sketch's structural validation and single intentional exception type are the right shape, and the problem you flag, that kryo.deserializeFrom runs the framework decode and user serializers with nothing between them, is exactly why I do not want to rush that boundary inside this review cycle. I will file that issue and link it here; it can also carry a strict mode that turns drops back into fatal errors if there is appetite for such a knob.

The minor points are both accepted and pushed. The try in recv() now covers des.deserialize only, so updateMetrics and the batch add run outside it and a post-decode failure propagates without being counted; testPostDecodeFailurePropagatesAndIsNotCounted pins this by throwing a tolerated-type exception from updateMetrics and asserting the exception escapes with the failure counter at zero. I have also rewritten the description to match the code: 8 types with no NullPointerException, the rate-limited logging, and the top-level deserializationFailures key.

The full storm-client suite is green locally (662 tests) with no checkstyle findings.

[L1nq0]

@reiabreu Filed as #9077, plus #9078 for the spout and offset interaction.

[rzo1]

lgtm

[rzo1]

Thanks for the contribution @L1nq0 and the reviews.