From 6ae03e42e53dfc465737c069c8454835707520f9 Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Sun, 11 Oct 2026 19:40:17 +0200 Subject: [PATCH 1/3] Make strict deserialization mode fatal for IOException-wrapped failures 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 #9157 --- docs/Metrics.md | 2 +- docs/Serialization.md | 2 +- .../src/jvm/org/apache/storm/Config.java | 6 +-- .../DeserializingConnectionCallback.java | 6 ++- .../FatalDeserializationException.java | 24 ++++++++++++ .../messaging/netty/StormServerHandler.java | 12 +++++- .../DeserializingConnectionCallbackTest.java | 29 ++++++++++++-- .../netty/StormServerHandlerTest.java | 39 +++++++++++++++++++ 8 files changed, 109 insertions(+), 11 deletions(-) create mode 100644 storm-client/src/jvm/org/apache/storm/messaging/FatalDeserializationException.java create mode 100644 storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java diff --git a/docs/Metrics.md b/docs/Metrics.md index 9a3a9fc83aa..433f9aae332 100644 --- a/docs/Metrics.md +++ b/docs/Metrics.md @@ -302,7 +302,7 @@ Be aware that the `__system` bolt is an actual bolt so regular bolt metrics desc `dequeuedMessages` is a throwback to older code where there was an internal queue between the server and the bolts/spouts. That is no longer the case and the value can be ignored. `enqueued` is a map between the address of the remote worker and the number of tuples that were sent from it to this worker. -`deserializationFailures` is the number of incoming messages that failed to deserialize and were dropped. When `topology.tuple.deserialization.strict.enable` is set, deserialization failures are not dropped or counted; they propagate instead. Non-IO failures terminate the worker, while failures caused by an `IOException`, such as a corrupt compressed frame, close the connection and leave the worker running. +`deserializationFailures` is the number of incoming messages that failed to deserialize and were dropped. When `topology.tuple.deserialization.strict.enable` is set, deserialization failures are not dropped or counted; they terminate the worker instead. ##### Send (Netty Client) diff --git a/docs/Serialization.md b/docs/Serialization.md index 4d3841ad896..714c234f3ba 100644 --- a/docs/Serialization.md +++ b/docs/Serialization.md @@ -131,7 +131,7 @@ Be aware that the topology-wide form enables compression for *every* remote-boun | `topology.tuple.compression.threshold` | `1460` | Minimum serialized tuple size, in bytes, before compression is attempted. Tuples at or below this size are sent uncompressed. The default matches the typical Ethernet TCP MSS, so payloads that already fit in a single network frame are never compressed. | | `storm.compression.zstd.level` | `3` | Zstd compression level. Supported range is 1–19; levels 20–22 (ultra mode) are prohibited because of their memory requirements. | | `topology.tuple.compression.max.decompressed.bytes` | `10485760` (10 MB) | Upper bound on the decompressed size of a single tuple. Decompression that would exceed this limit fails, guarding against malicious or corrupt payloads. | -| `topology.tuple.deserialization.strict.enable` | `false` | Makes tuple deserialization failures fatal: an incoming message that fails to decode kills the receiving worker instead of being dropped and counted. This restores the pre-3.1.0 behavior and is mainly useful for debugging, because a persistent bad frame will put the worker in a restart loop. A failure rooted in an `IOException`, such as a corrupt or oversized compressed frame, is an exception: the receiving worker closes that connection, discards the batch it was reading, and carries on while the sender reconnects. | +| `topology.tuple.deserialization.strict.enable` | `false` | Makes tuple deserialization failures fatal: an incoming message that fails to decode kills the receiving worker instead of being dropped and counted. This restores the pre-3.1.0 behavior and is mainly useful for debugging, because a persistent bad frame will put the worker in a restart loop. This includes failures rooted in an `IOException`, such as a corrupt or oversized compressed frame. | #### How decompression works diff --git a/storm-client/src/jvm/org/apache/storm/Config.java b/storm-client/src/jvm/org/apache/storm/Config.java index 9c275f82c40..9fec88235c5 100644 --- a/storm-client/src/jvm/org/apache/storm/Config.java +++ b/storm-client/src/jvm/org/apache/storm/Config.java @@ -1653,10 +1653,8 @@ public class Config extends HashMap { * By default a message that fails to decode on the receiving worker is dropped and counted, and the worker keeps * running. When set to {@code true}, any deserialization failure propagates and the worker exits, restoring the * pre-3.1.0 behavior. Be aware that a single corrupt frame from a peer then kills the worker, and the supervisor - * restarts it into the same failure, so a persistent bad frame results in a restart loop. Failures that wrap an - * {@code IOException}, such as a broken or oversized compressed frame, do not reach the exit path: the messaging - * layer treats them like a connection error, drops the batch being read, and leaves the worker running while the - * peer reconnects. + * restarts it into the same failure, so a persistent bad frame results in a restart loop. This includes failures + * that wrap an {@code IOException}, such as a broken or oversized compressed frame. * Default: {@code false}. */ @IsBoolean diff --git a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java index cee8776a1a4..b90830faf46 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java @@ -108,7 +108,11 @@ public void recv(List 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 + + " bytes destined for task " + message.task(), e); + } + if (!isToleratedDeserializationFailure(e)) { throw e; } deserializationFailures.incrementAndGet(); diff --git a/storm-client/src/jvm/org/apache/storm/messaging/FatalDeserializationException.java b/storm-client/src/jvm/org/apache/storm/messaging/FatalDeserializationException.java new file mode 100644 index 00000000000..ad6dae57317 --- /dev/null +++ b/storm-client/src/jvm/org/apache/storm/messaging/FatalDeserializationException.java @@ -0,0 +1,24 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.messaging; + +/** + * Wraps a tuple deserialization failure under {@link org.apache.storm.Config#TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE}. + * The messaging layer treats it as fatal even when the cause chain contains an exception it would otherwise tolerate, + * such as the {@code IOException} of a corrupt compressed frame. + */ +public class FatalDeserializationException extends RuntimeException { + public FatalDeserializationException(String message, Throwable cause) { + super(message, cause); + } +} diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java index 0088395d179..9bf0eedbc0e 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java @@ -14,8 +14,10 @@ import java.io.IOException; import java.util.Arrays; +import java.util.Collections; import java.util.HashSet; import java.util.Set; +import org.apache.storm.messaging.FatalDeserializationException; import org.apache.storm.shade.io.netty.channel.Channel; import org.apache.storm.shade.io.netty.channel.ChannelHandlerContext; import org.apache.storm.shade.io.netty.channel.ChannelInboundHandlerAdapter; @@ -59,7 +61,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { // Doing nothing (probably due to an oom issue) and hoping Utils.handleUncaughtException will handle it } try { - Utils.handleUncaughtException(cause, ALLOWED_EXCEPTIONS, false); + Utils.handleUncaughtException(cause, allowedExceptions(cause), false); ctx.close(); } catch (Error error) { LOG.info("Received error in netty thread.. terminating server..."); @@ -67,4 +69,12 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { } } + + // A strict-mode deserialization failure must terminate the worker even if it wraps an IOException. + static Set> allowedExceptions(Throwable cause) { + if (Utils.exceptionCauseIsInstanceOf(FatalDeserializationException.class, cause)) { + return Collections.emptySet(); + } + return ALLOWED_EXCEPTIONS; + } } diff --git a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java index de8bf62b291..a5e60c22125 100644 --- a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java +++ b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java @@ -170,7 +170,9 @@ public void testStrictModeMakesFailuresFatal() { WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); DeserializingConnectionCallback callback = new DeserializingConnectionCallback(conf, context, transfer); - assertThrows(KryoException.class, () -> callback.recv(Collections.singletonList(taskMessage(truncated)))); + FatalDeserializationException thrown = assertThrows(FatalDeserializationException.class, + () -> callback.recv(Collections.singletonList(taskMessage(truncated)))); + assertTrue(thrown.getCause() instanceof KryoException, "expected a KryoException cause but was: " + thrown.getCause()); verify(transfer, never()).transfer(any()); assertEquals(0L, callback.getAndResetDeserializationFailures()); @@ -184,8 +186,29 @@ public void testStrictModeMakesUnknownTaskFailureFatal() { WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); DeserializingConnectionCallback callback = new DeserializingConnectionCallback(conf, context, transfer); - assertThrows(TupleDeserializationException.class, - () -> callback.recv(Collections.singletonList(taskMessage(unknownSourceTaskTuple())))); + FatalDeserializationException thrown = assertThrows(FatalDeserializationException.class, + () -> callback.recv(Collections.singletonList(taskMessage(unknownSourceTaskTuple())))); + assertTrue(thrown.getCause() instanceof TupleDeserializationException, + "expected a TupleDeserializationException cause but was: " + thrown.getCause()); + + verify(transfer, never()).transfer(any()); + assertEquals(0L, callback.getAndResetDeserializationFailures()); + } + + @Test + public void testStrictModeMakesIoExceptionFailureFatal() { + Map conf = baseConf(); + conf.put(Config.TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE, true); + conf.put(Config.TOPOLOGY_TUPLE_COMPRESSION_ENABLE, true); + byte[] fakeZstd = {(byte) 0x28, (byte) 0xB5, (byte) 0x2F, (byte) 0xFD, 0x00, 0x01, 0x02, 0x03}; + + WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); + DeserializingConnectionCallback callback = new DeserializingConnectionCallback(conf, context, transfer); + + FatalDeserializationException thrown = assertThrows(FatalDeserializationException.class, + () -> callback.recv(Collections.singletonList(taskMessage(fakeZstd)))); + assertTrue(Utils.exceptionCauseIsInstanceOf(IOException.class, thrown), + "expected an IOException in the cause chain but was: " + thrown); verify(transfer, never()).transfer(any()); assertEquals(0L, callback.getAndResetDeserializationFailures()); diff --git a/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java b/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java new file mode 100644 index 00000000000..693c627b648 --- /dev/null +++ b/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java @@ -0,0 +1,39 @@ +/** + * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. The ASF licenses this file to you under the Apache License, Version + * 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions + * and limitations under the License. + */ + +package org.apache.storm.messaging.netty; + +import java.io.IOException; +import org.apache.storm.messaging.FatalDeserializationException; +import org.apache.storm.utils.Utils; +import org.junit.jupiter.api.Test; + +import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; +import static org.junit.jupiter.api.Assertions.assertThrows; + +public class StormServerHandlerTest { + + @Test + public void testIoExceptionIsSwallowed() { + RuntimeException cause = new RuntimeException("Failed to decompress", new IOException("corrupt frame")); + + assertDoesNotThrow(() -> Utils.handleUncaughtException(cause, StormServerHandler.allowedExceptions(cause), false)); + } + + @Test + public void testFatalDeserializationWrappingIoExceptionIsNotSwallowed() { + RuntimeException cause = new FatalDeserializationException("strict", + new RuntimeException("Failed to decompress", new IOException("corrupt frame"))); + + assertThrows(Error.class, () -> Utils.handleUncaughtException(cause, StormServerHandler.allowedExceptions(cause), false)); + } +} From a1ad9fda6397f391606f37b16f566ed4e38f8fb6 Mon Sep 17 00:00:00 2001 From: Richard Zowalla Date: Mon, 12 Oct 2026 20:53:59 +0200 Subject: [PATCH 2/3] Handle strict deserialization failures directly in exceptionCaught 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. --- .../messaging/netty/StormServerHandler.java | 30 ++++++++---- .../netty/StormServerHandlerTest.java | 48 +++++++++++++++---- 2 files changed, 59 insertions(+), 19 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java index 9bf0eedbc0e..b5df1e67713 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java @@ -14,7 +14,6 @@ import java.io.IOException; import java.util.Arrays; -import java.util.Collections; import java.util.HashSet; import java.util.Set; import org.apache.storm.messaging.FatalDeserializationException; @@ -29,9 +28,16 @@ public class StormServerHandler extends ChannelInboundHandlerAdapter { private static final Logger LOG = LoggerFactory.getLogger(StormServerHandler.class); private static final Set> ALLOWED_EXCEPTIONS = new HashSet<>(Arrays.asList(new Class[]{ IOException.class })); private final IServer server; + private final Runnable terminator; public StormServerHandler(IServer server) { + this(server, StormServerHandler::terminateWorker); + } + + // Package-private for testing. + StormServerHandler(IServer server, Runnable terminator) { this.server = server; + this.terminator = terminator; } @Override @@ -55,26 +61,32 @@ 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); + terminator.run(); + ctx.close(); + return; + } try { LOG.error("server errors in handling the request from {}", ctx.channel().remoteAddress().toString(), cause); } catch (Throwable err) { // Doing nothing (probably due to an oom issue) and hoping Utils.handleUncaughtException will handle it } try { - Utils.handleUncaughtException(cause, allowedExceptions(cause), false); + Utils.handleUncaughtException(cause, ALLOWED_EXCEPTIONS, false); ctx.close(); } catch (Error error) { LOG.info("Received error in netty thread.. terminating server..."); - Runtime.getRuntime().exit(1); + terminator.run(); } } - // A strict-mode deserialization failure must terminate the worker even if it wraps an IOException. - static Set> allowedExceptions(Throwable cause) { - if (Utils.exceptionCauseIsInstanceOf(FatalDeserializationException.class, cause)) { - return Collections.emptySet(); - } - return ALLOWED_EXCEPTIONS; + // 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"); + exitThread.start(); } } diff --git a/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java b/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java index 693c627b648..3caa9387d92 100644 --- a/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java +++ b/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java @@ -13,27 +13,55 @@ package org.apache.storm.messaging.netty; import java.io.IOException; +import java.net.InetSocketAddress; import org.apache.storm.messaging.FatalDeserializationException; -import org.apache.storm.utils.Utils; +import org.apache.storm.shade.io.netty.channel.Channel; +import org.apache.storm.shade.io.netty.channel.ChannelHandlerContext; +import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; -import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; -import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; public class StormServerHandlerTest { + private Runnable terminator; + private ChannelHandlerContext ctx; + private StormServerHandler handler; + + @BeforeEach + public void setUp() { + terminator = mock(Runnable.class); + Channel channel = mock(Channel.class); + when(channel.remoteAddress()).thenReturn(new InetSocketAddress("localhost", 6700)); + ctx = mock(ChannelHandlerContext.class); + when(ctx.channel()).thenReturn(channel); + handler = new StormServerHandler(mock(IServer.class), terminator); + } + + @Test + public void testIoExceptionClosesConnection() { + handler.exceptionCaught(ctx, new RuntimeException("Failed to decompress", new IOException("corrupt frame"))); + + verify(ctx).close(); + verify(terminator, never()).run(); + } + @Test - public void testIoExceptionIsSwallowed() { - RuntimeException cause = new RuntimeException("Failed to decompress", new IOException("corrupt frame")); + public void testFatalDeserializationWrappingIoExceptionTerminates() { + handler.exceptionCaught(ctx, new FatalDeserializationException("strict", + new RuntimeException("Failed to decompress", new IOException("corrupt frame")))); - assertDoesNotThrow(() -> Utils.handleUncaughtException(cause, StormServerHandler.allowedExceptions(cause), false)); + verify(terminator).run(); } @Test - public void testFatalDeserializationWrappingIoExceptionIsNotSwallowed() { - RuntimeException cause = new FatalDeserializationException("strict", - new RuntimeException("Failed to decompress", new IOException("corrupt frame"))); + public void testOtherFailureTerminates() { + handler.exceptionCaught(ctx, new IllegalStateException("unexpected")); - assertThrows(Error.class, () -> Utils.handleUncaughtException(cause, StormServerHandler.allowedExceptions(cause), false)); + verify(terminator).run(); + verify(ctx, never()).close(); } } From 44c0cb4e01dc255f48e060184b995511fb71975d Mon Sep 17 00:00:00 2001 From: Gianluca Graziadei Date: Tue, 13 Oct 2026 23:28:57 +0200 Subject: [PATCH 3/3] Address the review comments, moving reusable helper methods into Utils --- .../DeserializingConnectionCallback.java | 6 ++- .../apache/storm/messaging/netty/IServer.java | 2 + .../netty/NettyUncaughtExceptionHandler.java | 6 +-- .../apache/storm/messaging/netty/Server.java | 6 +++ .../messaging/netty/StormServerHandler.java | 26 ++++++++---- .../src/jvm/org/apache/storm/utils/Utils.java | 18 ++++++++ .../DeserializingConnectionCallbackTest.java | 15 +++++++ .../netty/StormServerHandlerTest.java | 41 ++++++++++++++++++- 8 files changed, 104 insertions(+), 16 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java index b90830faf46..3967ccc6deb 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/DeserializingConnectionCallback.java @@ -105,11 +105,13 @@ public void recv(List batch) { ArrayList ret = new ArrayList<>(batch.size()); for (TaskMessage message : batch) { Tuple tuple; + byte[] payload = message.message(); try { - tuple = des.deserialize(message.message()); + tuple = des.deserialize(payload); } catch (Exception e) { if (strictMode) { - throw new FatalDeserializationException("Failed to deserialize a message of " + message.message().length + int size = payload == null ? -1 : payload.length; + throw new FatalDeserializationException("Failed to deserialize a message of " + size + " bytes destined for task " + message.task(), e); } if (!isToleratedDeserializationFailure(e)) { diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/IServer.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/IServer.java index 4b095d68087..5f34365f740 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/netty/IServer.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/IServer.java @@ -18,4 +18,6 @@ public interface IServer { void channelActive(Channel c); void received(Object message, String remote, Channel channel) throws InterruptedException; + + default void terminating() {} } diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/NettyUncaughtExceptionHandler.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/NettyUncaughtExceptionHandler.java index e7bd779dd8b..a2d8998a75c 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/netty/NettyUncaughtExceptionHandler.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/NettyUncaughtExceptionHandler.java @@ -22,7 +22,7 @@ public class NettyUncaughtExceptionHandler implements Thread.UncaughtExceptionHa @Override public void uncaughtException(Thread t, Throwable e) { try { - LOG.error("Uncaught exception in netty " + e.getCause()); + LOG.error("Uncaught exception in netty", e); } catch (Throwable err) { // Doing nothing (probably due to an oom issue) and hoping Utils.handleUncaughtException will handle it } @@ -30,9 +30,9 @@ public void uncaughtException(Thread t, Throwable e) { try { Utils.handleUncaughtException(e); } catch (Throwable throwable) { - LOG.error("Exception thrown while handling uncaught exception " + throwable.getCause()); + LOG.error("Exception thrown while handling uncaught exception", throwable); } LOG.info("Received error in netty thread.. terminating server..."); - Runtime.getRuntime().exit(1); + Utils.exitProcessFromNewThread(1, "Uncaught exception in a netty thread"); } } diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java index 3d54cab0d7d..c1a790d681a 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/Server.java @@ -162,6 +162,7 @@ public int getPort() { */ @Override public void close() { + closing = true; allChannels.close().awaitUninterruptibly(); workerEventLoopGroup.shutdownGracefully().awaitUninterruptibly(); bossEventLoopGroup.shutdownGracefully().awaitUninterruptibly(); @@ -270,6 +271,11 @@ public void channelActive(Channel c) { allChannels.add(c); } + @Override + public void terminating() { + closing = true; + } + @Override public void received(Object message, String remote, Channel channel) throws InterruptedException { List msgs; diff --git a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java index b5df1e67713..3c8202711f7 100644 --- a/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java +++ b/storm-client/src/jvm/org/apache/storm/messaging/netty/StormServerHandler.java @@ -61,10 +61,9 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { - if (cause instanceof FatalDeserializationException) { + if (Utils.exceptionCauseIsInstanceOf(FatalDeserializationException.class, cause)) { // Strict deserialization mode: fatal even if the failure wraps an IOException. - LOG.error("Strict deserialization failure, terminating worker", cause); - terminator.run(); + terminate(ctx, cause, "Strict deserialization failure from {}, terminating worker"); ctx.close(); return; } @@ -77,16 +76,25 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) { Utils.handleUncaughtException(cause, ALLOWED_EXCEPTIONS, false); ctx.close(); } catch (Error error) { - LOG.info("Received error in netty thread.. terminating server..."); - terminator.run(); + terminate(ctx, cause, "Received error in netty thread from {}, terminating server..."); } } - // 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. + // Stops the server delivering further batches (the event loops run until the exit completes), then exits. + // Logging comes last and is guarded: a failure to log (e.g. an OOM) must not skip the termination. + private void terminate(ChannelHandlerContext ctx, Throwable cause, String message) { + server.terminating(); + terminator.run(); + try { + LOG.error(message, ctx.channel().remoteAddress(), cause); + } catch (Throwable err) { + // nop + } + } + + // Not on this thread: the worker shutdown hook closes the netty server, which waits for this event loop. private static void terminateWorker() { - Thread exitThread = new Thread(() -> Runtime.getRuntime().exit(1), "netty-server-exit"); - exitThread.start(); + Utils.exitProcessFromNewThread(1, "Fatal error in the netty server"); } } diff --git a/storm-client/src/jvm/org/apache/storm/utils/Utils.java b/storm-client/src/jvm/org/apache/storm/utils/Utils.java index feebf0fabaf..cd0ffdeb4c6 100644 --- a/storm-client/src/jvm/org/apache/storm/utils/Utils.java +++ b/storm-client/src/jvm/org/apache/storm/utils/Utils.java @@ -67,6 +67,7 @@ import java.util.TreeMap; import java.util.UUID; import java.util.concurrent.Callable; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.jar.JarFile; import java.util.regex.Matcher; import java.util.regex.Pattern; @@ -121,6 +122,7 @@ public class Utils { public static final Logger LOG = LoggerFactory.getLogger(Utils.class); + private static final AtomicBoolean EXIT_REQUESTED = new AtomicBoolean(); public static final String DEFAULT_STREAM_ID = "default"; private static final Set> defaultAllowedExceptions = Collections.emptySet(); private static final List LOCALHOST_ADDRESSES = Lists.newArrayList("localhost", "127.0.0.1", "0:0:0:0:0:0:0:1"); @@ -528,6 +530,22 @@ public static void exitProcess(int val, String msg) { Runtime.getRuntime().exit(val); } + /** + * Exits the process like {@link #exitProcess(int, String)}, but from a new thread. Needed when the caller runs on + * a thread that a shutdown hook waits for, such as a Netty event loop: the worker shutdown hook closes the Netty + * server, which awaits its event loops, while {@code Runtime.exit()} blocks the calling thread until the hooks + * complete. Idempotent: once an exit is in progress, further calls return immediately instead of starting another + * thread that would block in {@code Runtime.exit()} forever. + */ + public static void exitProcessFromNewThread(int val, String msg) { + if (!EXIT_REQUESTED.compareAndSet(false, true)) { + return; + } + Thread exitThread = new Thread(() -> exitProcess(val, msg), "process-exit"); + exitThread.setDaemon(false); + exitThread.start(); + } + public static String uuid() { return UUID.randomUUID().toString(); } diff --git a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java index a5e60c22125..fbc5d60615b 100644 --- a/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java +++ b/storm-client/test/jvm/org/apache/storm/messaging/DeserializingConnectionCallbackTest.java @@ -178,6 +178,21 @@ public void testStrictModeMakesFailuresFatal() { assertEquals(0L, callback.getAndResetDeserializationFailures()); } + @Test + public void testStrictModeReportsNullPayloadAsFatal() { + Map conf = baseConf(); + conf.put(Config.TOPOLOGY_TUPLE_DESERIALIZATION_STRICT_ENABLE, true); + + WorkerState.ILocalTransferCallback transfer = mock(WorkerState.ILocalTransferCallback.class); + DeserializingConnectionCallback callback = new DeserializingConnectionCallback(conf, context, transfer); + + FatalDeserializationException thrown = assertThrows(FatalDeserializationException.class, + () -> callback.recv(Collections.singletonList(taskMessage(null)))); + assertTrue(thrown.getMessage().contains("-1 bytes"), "unexpected message: " + thrown.getMessage()); + + verify(transfer, never()).transfer(any()); + } + @Test public void testStrictModeMakesUnknownTaskFailureFatal() { Map conf = baseConf(); diff --git a/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java b/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java index 3caa9387d92..bf8d281bb04 100644 --- a/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java +++ b/storm-client/test/jvm/org/apache/storm/messaging/netty/StormServerHandlerTest.java @@ -28,23 +28,36 @@ public class StormServerHandlerTest { private Runnable terminator; + private IServer server; + private Channel channel; private ChannelHandlerContext ctx; private StormServerHandler handler; @BeforeEach public void setUp() { terminator = mock(Runnable.class); - Channel channel = mock(Channel.class); + server = mock(IServer.class); + channel = mock(Channel.class); when(channel.remoteAddress()).thenReturn(new InetSocketAddress("localhost", 6700)); ctx = mock(ChannelHandlerContext.class); when(ctx.channel()).thenReturn(channel); - handler = new StormServerHandler(mock(IServer.class), terminator); + handler = new StormServerHandler(server, terminator); } @Test public void testIoExceptionClosesConnection() { handler.exceptionCaught(ctx, new RuntimeException("Failed to decompress", new IOException("corrupt frame"))); + verify(ctx).close(); + verify(terminator, never()).run(); + verify(server, never()).terminating(); + } + + @Test + public void testSocketIoExceptionClosesConnection() { + // what netty itself reports for a connection error: a bare IOException + handler.exceptionCaught(ctx, new IOException("Connection reset by peer")); + verify(ctx).close(); verify(terminator, never()).run(); } @@ -54,13 +67,37 @@ public void testFatalDeserializationWrappingIoExceptionTerminates() { handler.exceptionCaught(ctx, new FatalDeserializationException("strict", new RuntimeException("Failed to decompress", new IOException("corrupt frame")))); + verify(server).terminating(); + verify(terminator).run(); + verify(ctx).close(); + } + + @Test + public void testWrappedFatalDeserializationTerminates() { + handler.exceptionCaught(ctx, new RuntimeException("rethrown", + new FatalDeserializationException("strict", new IOException("corrupt frame")))); + + verify(server).terminating(); + verify(terminator).run(); + verify(ctx).close(); + } + + @Test + public void testFatalDeserializationTerminatesEvenIfLoggingFails() { + when(channel.remoteAddress()).thenThrow(new OutOfMemoryError("no heap left to log")); + + handler.exceptionCaught(ctx, new FatalDeserializationException("strict", new IOException("corrupt frame"))); + + verify(server).terminating(); verify(terminator).run(); + verify(ctx).close(); } @Test public void testOtherFailureTerminates() { handler.exceptionCaught(ctx, new IllegalStateException("unexpected")); + verify(server).terminating(); verify(terminator).run(); verify(ctx, never()).close(); }