GGraziadei commented on code in PR #9159:
URL: https://github.com/apache/storm/pull/9159#discussion_r4261837979
##########
storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java:
##########
@@ -53,6 +61,13 @@ public void channelRead(ChannelHandlerContext ctx, Object
msg) throws Exception
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
+ if (cause instanceof FatalDeserializationException) {
Review 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.
##########
storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java:
##########
@@ -53,6 +61,13 @@ public void channelRead(ChannelHandlerContext ctx, Object
msg) throws Exception
@Override
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);
Review 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
}
```
##########
storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java:
##########
@@ -63,8 +78,15 @@ public void exceptionCaught(ChannelHandlerContext ctx,
Throwable cause) {
ctx.close();
} catch (Error error) {
LOG.info("Received error in netty thread.. terminating server...");
- Runtime.getRuntime().exit(1);
+ terminator.run();
}
}
+
+ // 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");
Review 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;
}
```
##########
storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java:
##########
@@ -108,7 +108,11 @@ public void recv(List<TaskMessage> batch) {
try {
tuple = des.deserialize(message.message());
} catch (Exception e) {
- if (strictMode || !isToleratedDeserializationFailure(e)) {
+ if (strictMode) {
+ throw new FatalDeserializationException("Failed to
deserialize a message of " + message.message().length
Review 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);
}
...
}
```
--
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]
