0

我们有一个使用 3 个节点和多个拓扑运行的风暴集群。我们apache-storm-1.2.2使用java 1.8.0_162

目前我们遇到的问题是随机拓扑在发生错误并且 Netty 服务器不可用后随机停止发射。这可能会在几个小时或几天后发生。

由于我们没有改变风暴螺栓发射或执行数据的逻辑，我们目前不知道如何抛出这样的错误。同样值得怀疑的是，为什么在这样的错误之后整个拓扑停止工作。

似乎某些 HashMap 的反序列化存在问题。但我们无法弄清楚这是怎么发生的。

这里是导致失败的一名工人的错误：

2019-09-24 14:31:02.414 o.a.s.m.n.StormServerHandler Netty-server-localhost-6727-worker-2 [ERROR] server errors in handling the request
java.lang.RuntimeException: java.io.OptionalDataException
        at org.apache.storm.serialization.SerializableSerializer.read(SerializableSerializer.java:58) ~[storm-core-1.2.2.jar:1.2.2]
        at com.esotericsoftware.kryo.Kryo.readClassAndObject(Kryo.java:793) ~[kryo-3.0.3.jar:?]
        at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:134) ~[kryo-3.0.3.jar:?]
        at com.esotericsoftware.kryo.serializers.CollectionSerializer.read(CollectionSerializer.java:40) ~[kryo-3.0.3.jar:?]
        at com.esotericsoftware.kryo.Kryo.readObject(Kryo.java:689) ~[kryo-3.0.3.jar:?]
        at org.apache.storm.serialization.KryoValuesDeserializer.deserializeFrom(KryoValuesDeserializer.java:37) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.serialization.KryoTupleDeserializer.deserialize(KryoTupleDeserializer.java:50) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.messaging.DeserializingConnectionCallback.recv(DeserializingConnectionCallback.java:56) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.messaging.netty.Server.enqueue(Server.java:134) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.messaging.netty.Server.received(Server.java:255) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.messaging.netty.StormServerHandler.messageReceived(StormServerHandler.java:61) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) ~[storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.DefaultChannelPipeline$DefaultChannelHandlerContext.sendUpstream(DefaultChannelPipeline.java:791) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:296) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.handler.codec.frame.FrameDecoder.unfoldAndFireMessageReceived(FrameDecoder.java:462) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.handler.codec.frame.FrameDecoder.callDecode(FrameDecoder.java:443) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.handler.codec.frame.FrameDecoder.messageReceived(FrameDecoder.java:310) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.SimpleChannelUpstreamHandler.handleUpstream(SimpleChannelUpstreamHandler.java:70) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:564) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.DefaultChannelPipeline.sendUpstream(DefaultChannelPipeline.java:559) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:268) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.Channels.fireMessageReceived(Channels.java:255) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.socket.nio.NioWorker.read(NioWorker.java:88) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.socket.nio.AbstractNioWorker.process(AbstractNioWorker.java:108) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.socket.nio.AbstractNioSelector.run(AbstractNioSelector.java:337) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.socket.nio.AbstractNioWorker.run(AbstractNioWorker.java:89) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.channel.socket.nio.NioWorker.run(NioWorker.java:178) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.util.ThreadRenamingRunnable.run(ThreadRenamingRunnable.java:108) [storm-core-1.2.2.jar:1.2.2]
        at org.apache.storm.shade.org.jboss.netty.util.internal.DeadLockProofWorker$1.run(DeadLockProofWorker.java:42) [storm-core-1.2.2.jar:1.2.2]
        at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149) [?:1.8.0_162]
        at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624) [?:1.8.0_162]
        at java.lang.Thread.run(Thread.java:748) [?:1.8.0_162]
Caused by: java.io.OptionalDataException
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1587) ~[?:1.8.0_162]
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:427) ~[?:1.8.0_162]
        at java.util.HashMap.readObject(HashMap.java:1407) ~[?:1.8.0_162]
        at sun.reflect.GeneratedMethodAccessor34.invoke(Unknown Source) ~[?:?]
        at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_162]
        at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_162]
        at java.io.ObjectStreamClass.invokeReadObject(ObjectStreamClass.java:1158) ~[?:1.8.0_162]
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2169) ~[?:1.8.0_162]
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2060) ~[?:1.8.0_162]
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1567) ~[?:1.8.0_162]
        at java.io.ObjectInputStream.defaultReadFields(ObjectInputStream.java:2278) ~[?:1.8.0_162]
        at java.io.ObjectInputStream.readSerialData(ObjectInputStream.java:2202) ~[?:1.8.0_162]
        at java.io.ObjectInputStream.readOrdinaryObject(ObjectInputStream.java:2060) ~[?:1.8.0_162]
        at java.io.ObjectInputStream.readObject0(ObjectInputStream.java:1567) ~[?:1.8.0_162]
        at java.io.ObjectInputStream.readObject(ObjectInputStream.java:427) ~[?:1.8.0_162]
        at org.apache.storm.serialization.SerializableSerializer.read(SerializableSerializer.java:56) ~[storm-core-1.2.2.jar:1.2.2]
        ... 32 more

抛出此错误后，此特定拓扑的所有其他工作人员将停止工作，我们必须杀死并重新部署。

其他工作人员的日志要么显示无用，要么如下：

2019-09-24 14:31:03.012 o.a.s.m.n.Client Thread-32-disruptor-worker-transfer-queue [ERROR] connection to Netty-Client-HOSTE_NAME/IP:PORT is unavailable
2019-09-24 14:31:03.053 o.a.s.m.n.Client client-worker-1 [WARN] Re-connection to HOSTE_NAME/IP:PORT was successful but 4 messages has been lost so far
4

1 回答 1

0

根据您发布的内容，以下几点可以帮助您缩小问题范围：

关于您的拓扑空闲，您应该尝试至少升级到 Storm 1.2.3。您可能会受到https://issues.apache.org/jira/browse/STORM-3055的影响。

您的堆栈跟踪显示您正在对某些元组使用 Java 序列化。我认为这里发生的事情是您的一个工作人员已经序列化了一个元组，该元组被发送给另一个工作人员，然后该元组的反序列化失败。请参阅https://docs.oracle.com/javase/7/docs/api/java/io/OptionalDataException.html。

首先，我认为您应该尝试关闭 Java 序列化。它很慢，而且您可能无论如何都不希望它在生产拓扑中使用。请参阅https://storm.apache.org/releases/1.2.3/Serialization.html。

其次，下次您的拓扑挂起时（假设升级到 1.2.3 无法修复它），我会尝试使用jstack为空闲工作人员获取线程转储。这应该告诉你为什么拓扑没有做任何事情。

最后，您可以尝试在https://github.com/apache/storm/blob/5b01c78d053b2b8e2b7e6a4a8acd4c822924bbf9/storm-core/src/jvm/org/apache/storm/serialization/SerializableSerializer.java#L51临时记录字节数组当您看到异常时，他们会告诉您出了什么问题。您得到的异常表明字节数组不包含序列化 HashMap 的预期数据。要添加此日志记录，您需要编辑和构建 Storm，请参阅https://github.com/apache/storm/blob/master/DEVELOPER.md#building了解如何执行此操作。

于 2019-09-24T20:12:25.357 回答