Doppler Client appears to be failing on large messages from firehose

[bijukunjummen]

I don't have a good sample test for this - on a standalone (https://github.com/bijukunjummen/boot-firehose-nozzle-sample) test cf doppler client org.cloudfoundry.doppler.DopplerClient listening to a websocket feed from firehose, if a large websocket packet comes through then it fails with this exception:

reactor.core.util.Exceptions$UpstreamException: java.io.EOFException
    at reactor.core.util.Exceptions.bubble(Exceptions.java:112) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.util.Exceptions.onErrorDropped(Exceptions.java:160) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.subscriber.LambdaSubscriber.doError(LambdaSubscriber.java:144) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.subscriber.LambdaSubscriber.onError(LambdaSubscriber.java:131) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:137) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.publisher.FluxMap$MapSubscriber.onError(FluxMap.java:137) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.publisher.FluxFilter$FilterSubscriber.onError(FluxFilter.java:157) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onError(FluxMap.java:273) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onError(FluxMap.java:273) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:222) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.publisher.MonoFlatMap$FlattenSubscriber$InnerSubscriber.onNext(MonoFlatMap.java:195) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:125) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.core.publisher.FluxMap$MapSubscriber.onNext(FluxMap.java:125) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    at reactor.io.netty.common.NettyChannelHandler$ChannelInputSubscriber.onNext(NettyChannelHandler.java:426) ~[reactor-netty-2.5.0.BUILD-20160610.230521-329.jar:na]
    at reactor.io.netty.common.NettyChannelHandler.doRead(NettyChannelHandler.java:207) ~[reactor-netty-2.5.0.BUILD-20160610.230521-329.jar:na]
    at reactor.io.netty.http.NettyWebSocketClientHandler.channelRead(NettyWebSocketClientHandler.java:135) ~[reactor-netty-2.5.0.BUILD-20160610.230521-329.jar:na]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:326) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:293) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.fireChannelRead(ByteToMessageDecoder.java:280) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:396) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:326) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.ChannelInboundHandlerAdapter.channelRead(ChannelInboundHandlerAdapter.java:86) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:326) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.handler.ssl.SslHandler.unwrap(SslHandler.java:1070) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.handler.ssl.SslHandler.decode(SslHandler.java:904) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:411) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:248) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:326) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1320) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:334) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:905) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:123) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:563) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:504) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:418) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:390) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:742) [netty-all-4.1.1.Final.jar:4.1.1.Final]
    at java.lang.Thread.run(Thread.java:745) [na:1.8.0_77]
Caused by: java.io.EOFException: null
    at okio.RealBufferedSource.require(RealBufferedSource.java:64) ~[okio-1.5.0.jar:na]
    at com.squareup.wire.ProtoReader.beforeLengthDelimitedScalar(ProtoReader.java:387) ~[wire-runtime-2.1.2.jar:na]
    at com.squareup.wire.ProtoReader.readBytes(ProtoReader.java:258) ~[wire-runtime-2.1.2.jar:na]
    at com.squareup.wire.ProtoAdapter$13.decode(ProtoAdapter.java:378) ~[wire-runtime-2.1.2.jar:na]
    at com.squareup.wire.ProtoAdapter$13.decode(ProtoAdapter.java:368) ~[wire-runtime-2.1.2.jar:na]
    at org.cloudfoundry.dropsonde.events.LogMessage$ProtoAdapter_LogMessage.decode(LogMessage.java:310) ~[cloudfoundry-client-2.0.0.BUILD-20160616.144138-933.jar:2.0.0.BUILD-SNAPSHOT]
    at org.cloudfoundry.dropsonde.events.LogMessage$ProtoAdapter_LogMessage.decode(LogMessage.java:277) ~[cloudfoundry-client-2.0.0.BUILD-20160616.144138-933.jar:2.0.0.BUILD-SNAPSHOT]
    at org.cloudfoundry.dropsonde.events.Envelope$ProtoAdapter_Envelope.decode(Envelope.java:539) ~[cloudfoundry-client-2.0.0.BUILD-20160616.144138-933.jar:2.0.0.BUILD-SNAPSHOT]
    at org.cloudfoundry.dropsonde.events.Envelope$ProtoAdapter_Envelope.decode(Envelope.java:471) ~[cloudfoundry-client-2.0.0.BUILD-20160616.144138-933.jar:2.0.0.BUILD-SNAPSHOT]
    at com.squareup.wire.ProtoAdapter.decode(ProtoAdapter.java:182) ~[wire-runtime-2.1.2.jar:na]
    at com.squareup.wire.ProtoAdapter.decode(ProtoAdapter.java:176) ~[wire-runtime-2.1.2.jar:na]
    at org.cloudfoundry.reactor.doppler.ReactorDopplerClient.toEnvelope(ReactorDopplerClient.java:96) ~[cloudfoundry-client-spring-2.0.0.BUILD-20160616.144138-934.jar:2.0.0.BUILD-SNAPSHOT]
    at reactor.core.publisher.FluxMap$MapConditionalSubscriber.onNext(FluxMap.java:218) ~[reactor-core-2.5.0.BUILD-20160613.093553-691.jar:na]
    ... 38 common frames omitted

[cf-gitbot]

We have created an issue in Pivotal Tracker to manage this:

https://www.pivotaltracker.com/story/show/121742537

The labels on this github issue will be updated when the story is started.

[bijukunjummen]

@smaldini, this appears to be related to the reactor websocket client implementation - I tried debugging this but did not get very far. Let me know if you want me to show the behavior to you in some way.

[nebhale]

@bijukunjummen The reactor network has been reworked pretty significantly recently. Can you give another go with the most recent snapshot and see if you're still experiencing the problem?

[bijukunjummen]

Confirming that this works cleanly now @nebhale

[ctatineni]

I am facing the similar issue , i am using cloudfoundry-client-reactor 2.0.0.RELEASE, is the fix included in this release or do i have to upgrade reactor core and reactor netty jars?

[nebhale]

We believe that we've finally gotten to the bottom of this and it will be fixed in 2.1.0.RELEASE, once and for all.