reiabreu commented on PR #9076:
URL: https://github.com/apache/storm/pull/9076#issuecomment-5560846714
_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 #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 your #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:
```java
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:
```java
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:
```java
} 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 — the tests and the live-cluster check are
strong.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
