拉谷歌 PubSub 消息时 io.grpc 帧大小超过最大值

Posted 2023-02-16

【中文标题】拉谷歌 PubSub 消息时 io.grpc 帧大小超过最大值

【英文标题】：io.grpc frame size exceeds maximum when pull google PubSub message

【发布时间】：2017-04-30 06:06:57

【问题描述】：

有人知道在使用 Google PubSub 服务时如何增加 io.grpc maxMessageSize 吗？

现在，当我拉出一条大消息时，有时会收到此异常：

Exception in thread "main" com.google.cloud.pubsub.PubSubException: io.grpc.StatusRuntimeException: INTERNAL: Exception deframing message
    at com.google.cloud.pubsub.spi.DefaultPubSubRpc$1.apply(DefaultPubSubRpc.java:193)
    at com.google.cloud.pubsub.spi.DefaultPubSubRpc$1.apply(DefaultPubSubRpc.java:187)
    at com.google.common.util.concurrent.Futures$CatchingFuture.doFallback(Futures.java:842)
    at com.google.common.util.concurrent.Futures$CatchingFuture.doFallback(Futures.java:834)
    at com.google.common.util.concurrent.Futures$AbstractCatchingFuture.run(Futures.java:789)
    at com.google.common.util.concurrent.MoreExecutors$DirectExecutor.execute(MoreExecutors.java:456)
    at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:817)
    at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:753)
    at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:634)
    at com.google.api.gax.grpc.RetryingCallable$RetryingResultFuture.setException(RetryingCallable.java:202)
    at com.google.api.gax.grpc.RetryingCallable$Retryer.onFailure(RetryingCallable.java:147)
    at com.google.common.util.concurrent.Futures$6.run(Futures.java:1764)
    at com.google.common.util.concurrent.MoreExecutors$DirectExecutor.execute(MoreExecutors.java:456)
    at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:817)
    at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:753)
    at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:634)
    at com.google.api.gax.grpc.ExceptionTransformingCallable$ExceptionTransformingFuture.onFailure(ExceptionTransformingCallable.java:113)
    at com.google.common.util.concurrent.Futures$6.run(Futures.java:1764)
    at com.google.common.util.concurrent.MoreExecutors$DirectExecutor.execute(MoreExecutors.java:456)
    at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:817)
    at com.google.common.util.concurrent.AbstractFuture.complete(AbstractFuture.java:753)
    at com.google.common.util.concurrent.AbstractFuture.setException(AbstractFuture.java:634)
    at io.grpc.stub.ClientCalls$GrpcFuture.setException(ClientCalls.java:466)
    at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:442)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.close(ClientCallImpl.java:481)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.access$600(ClientCallImpl.java:398)
    at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:513)
    at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:52)
    at io.grpc.internal.SerializingExecutor$TaskRunner.run(SerializingExecutor.java:154)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
    at java.util.concurrent.FutureTask.run(FutureTask.java:266)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
    at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
    at java.lang.Thread.run(Thread.java:745)
Caused by: com.google.api.gax.grpc.ApiException:        io.grpc.StatusRuntimeException: INTERNAL: Exception deframing message
    ... 20 more
Caused by: io.grpc.StatusRuntimeException: INTERNAL: Exception deframing message
    at io.grpc.Status.asRuntimeException(Status.java:545)
    ... 13 more
Caused by: io.grpc.StatusRuntimeException: INTERNAL: Frame size 5454271 exceeds maximum: 4194304. If this is normal, increase the maxMessageSize in the channel/server builder
    at io.grpc.Status.asRuntimeException(Status.java:536)
    at io.grpc.internal.MessageDeframer.processHeader(MessageDeframer.java:338)
    at io.grpc.internal.MessageDeframer.deliver(MessageDeframer.java:240)
    at io.grpc.internal.MessageDeframer.deframe(MessageDeframer.java:176)
    at io.grpc.internal.AbstractStream.deframe(AbstractStream.java:276)
    at io.grpc.internal.AbstractClientStream.inboundDataReceived(AbstractClientStream.java:150)
    at io.grpc.internal.Http2ClientStream.transportDataReceived(Http2ClientStream.java:137)
    at io.grpc.netty.NettyClientStream.transportDataReceived(NettyClientStream.java:180)
    at io.grpc.netty.NettyClientHandler.onDataRead(NettyClientHandler.java:251)
    at io.grpc.netty.NettyClientHandler.access$700(NettyClientHandler.java:92)
    at io.grpc.netty.NettyClientHandler$FrameListener.onDataRead(NettyClientHandler.java:588)
    at io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder$FrameReadListener.onDataRead(DefaultHttp2ConnectionDecoder.java:256)
    at io.netty.handler.codec.http2.Http2InboundFrameLogger$1.onDataRead(Http2InboundFrameLogger.java:48)
    at io.netty.handler.codec.http2.DefaultHttp2FrameReader.readDataFrame(DefaultHttp2FrameReader.java:410)
    at io.netty.handler.codec.http2.DefaultHttp2FrameReader.processPayloadState(DefaultHttp2FrameReader.java:244)
    at io.netty.handler.codec.http2.DefaultHttp2FrameReader.readFrame(DefaultHttp2FrameReader.java:155)
    at io.netty.handler.codec.http2.Http2InboundFrameLogger.readFrame(Http2InboundFrameLogger.java:41)
    at io.netty.handler.codec.http2.DefaultHttp2ConnectionDecoder.decodeFrame(DefaultHttp2ConnectionDecoder.java:113)
    at io.netty.handler.codec.http2.Http2ConnectionHandler$FrameDecoder.decode(Http2ConnectionHandler.java:333)
    at io.netty.handler.codec.http2.Http2ConnectionHandler.decode(Http2ConnectionHandler.java:393)
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:372)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:358)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:350)
    at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1066)
    at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:900)
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411)
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:372)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:358)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:350)
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1334)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:372)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:358)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:926)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:123)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:571)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:512)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:426)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:398)
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:877)
    at io.netty.util.concurrent.DefaultThreadFactory$DefaultRunnableDecorator.run(DefaultThreadFactory.java:144)
    ... 1 more

该应用是用 java 编写的，使用 http://googlecloudplatform.github.io/google-cloud-java/0.6.0/apidocs/index.html?com/google/cloud/pubsub/PubSub.html 并在 Google Compute Engine 实例中运行。

【参考方案1】：

您可以通过NettyChannelBuilder.maxMessageSize() 之类的方式配置频道。

在 grpc v1.1 中，您将能够使用 ManagedChannelBuilder.maxInboundMessageSize() 并避免直接使用 NettyChannelBuilder。您还可以在 CallOptions 和 AbstractStub 上使用 withMaxInboundMessageSize() 对每个呼叫进行配置。