Repository navigation
Conversation
With topology.tuple.deserialization.strict.enable set, a decode failure is rethrown from DeserializingConnectionCallback.recv and reaches StormServerHandler.exceptionCaught, which swallows anything with an IOException in its cause chain. Failures from a corrupt or oversized compressed frame wrap an IOException, so they closed the connection and dropped the batch instead of terminating the worker. Strict-mode failures are now wrapped in FatalDeserializationException, and StormServerHandler does not apply the IOException exemption to it. Genuine connection errors keep the existing handling. Fixes apache#9157
rzo1
force-pushed
the
9157-strict-mode-io-failures
branch
from
October 11, 2026 17:41
6088ef9 to
6ae03e4
Compare
rzo1
marked this pull request as ready for review
October 11, 2026 17:41
rzo1
requested review from
GGraziadei and
reiabreu
and
a balanced review from Copilot
and removed request for
Copilot
October 11, 2026 17:42
Copilot stopped reviewing on behalf of
rzo1 due to an error
October 11, 2026 17:43
GGraziadei
reviewed
Oct 12, 2026
GGraziadei
left a comment
GGraziadei
left a comment
Member
There was a problem hiding this comment.
Hi @rzo1, thanks!
Just some small changes required. Some of them identified with the use of Opus, others manually.
FatalDeserializationException is now checked at the top of StormServerHandler.exceptionCaught instead of going through an empty allowed-exception set and the catch (Error) block. The worker is terminated from a separate thread. Calling exit() on the netty event loop thread blocks forever, because the worker shutdown hook closes the netty server and waits for that same event loop to finish. The tests now call exceptionCaught itself, with the terminator injected.
GGraziadei
requested changes
Oct 13, 2026
GGraziadei
left a comment
GGraziadei
left a comment
Member
There was a problem hiding this comment.
Hi @rzo1 thanks, some changes required.
|
|
||
| @Override | ||
| public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { | ||
| if (cause instanceof FatalDeserializationException) { |
Member
There was a problem hiding this comment.
Hi @rzo1 what do you think about using the Utils API?
Utils.exceptionCauseIsInstanceOf(FatalDeserializationException.class, cause)
This API checks on the whole chain, internally has an unwrap method.
| public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { | ||
| if (cause instanceof FatalDeserializationException) { | ||
| // Strict deserialization mode: fatal even if the failure wraps an IOException. | ||
| LOG.error("Strict deserialization failure, terminating worker", cause); |
Member
There was a problem hiding this comment.
What about extending the diagnostic in this way?
try {
LOG.error("Strict deserialization failure from {}, terminating worker",
ctx.channel().remoteAddress(), cause);
} catch (Throwable err) {
// nop
}
| // Exit from a separate thread: the worker shutdown hook closes the netty server, which waits for this | ||
| // event loop to finish, so calling exit() on the event loop thread itself would never return. | ||
| private static void terminateWorker() { | ||
| Thread exitThread = new Thread(() -> Runtime.getRuntime().exit(1), "netty-server-exit"); |
Member
There was a problem hiding this comment.
What about adding an atomic barrier to prevent multiple termination threads from being spawned?
A CAS operation could be used coupled with an atomic isTerminating flag.
if (! isTerminating.compareAndSet(false, true)) {
return;
}
| } catch (Exception e) { | ||
| if (strictMode || !isToleratedDeserializationFailure(e)) { | ||
| if (strictMode) { | ||
| throw new FatalDeserializationException("Failed to deserialize a message of " + message.message().length |
Member
There was a problem hiding this comment.
Please prevent NPE on message.message()
byte[] payload = message.message();
try {
// deserialize the payload ...
} catch (Exception e) {
if (strictMode) {
int size = payload == null ? -1 : payload.length;
throw new FatalDeserializationException("Failed to deserialize a message of " + size
+ " bytes destined for task " + message.task(), e);
}
...
}
Contributor
Author
|
Maybe just push your changes on the branch @GGraziadei ;-) |
Contributor
Author
|
Thanks @GGraziadei ! Changes lgtm |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Fixes #9157.
Follow-up to #9094.
Problem
In strict mode,
DeserializingConnectionCallback.recvrethrows the decode failure, and it reachesStormServerHandler.exceptionCaught. That method callsUtils.handleUncaughtExceptionwithIOExceptionas the only allowed exception, and the check walks the cause chain.ZstdUtils.decompressandKryoTupleDeserializer.deserializeTupleboth wrap anIOException, so a corrupt or oversized compressed frame closed the connection and discarded the batch, and the worker kept running.Change
DeserializingConnectionCallbackwraps the failure in a newFatalDeserializationException, keeping the original as the cause.StormServerHandlerdoes not apply theIOExceptionexemption when aFatalDeserializationExceptionis in the chain, so every strict-mode failure terminates the worker. Genuine connectionIOExceptions are handled as before.Configjavadoc,docs/Serialization.mdanddocs/Metrics.mdno longer describe the IOException case as an exception to strict mode.deserializationFailuresis still not incremented on the strict path, since the worker exits.Tests
DeserializingConnectionCallbackTest: the strict-mode tests expectFatalDeserializationExceptionwith the original cause, plus a new strict-mode test for a corrupt zstd frame.StormServerHandlerTest: a plain IOException-wrapping failure is swallowed, and one wrapped inFatalDeserializationExceptionis not.