Persistent subscription - Connection Closed

[jselamy]

We are using this library in production.

We have an event handler that dispatches to downstream apps subscribed to a stream. We have a stream in particular that has a huge load of incoming event (IoT)

After a long period of time, for some reasons looks like we're dealing with an issue whereas it's not responding anymore any idea why ?

We are facing this issue for quitte a time now.

[jselamy]

Any news ?

This is the error we get after a period of time on an active Persistent Subscription streaming events constantly

com.github.msemys.esjc.ConnectionClosedException: Connection was closed.
	at com.github.msemys.esjc.subscription.AbstractSubscriptionOperation.connectionClosed(AbstractSubscriptionOperation.java:219) ~[esjc-2.3.0.jar:2.3.0]
	at com.github.msemys.esjc.subscription.manager.SubscriptionManager.lambda$purgeSubscribedAndDropped$2(SubscriptionManager.java:55) ~[esjc-2.3.0.jar:2.3.0]
	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183) ~[na:na]
	at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:177) ~[na:na]
	at java.base/java.util.concurrent.ConcurrentHashMap$ValueSpliterator.forEachRemaining(ConcurrentHashMap.java:3612) ~[na:na]
	at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484) ~[na:na]
	at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474) ~[na:na]
	at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150) ~[na:na]
	at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173) ~[na:na]
	at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) ~[na:na]
	at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497) ~[na:na]
	at com.github.msemys.esjc.subscription.manager.SubscriptionManager.purgeSubscribedAndDropped(SubscriptionManager.java:54) ~[esjc-2.3.0.jar:2.3.0]
	at com.github.msemys.esjc.EventStoreTcp.onTcpConnectionClosed(EventStoreTcp.java:818) ~[esjc-2.3.0.jar:2.3.0]
	at com.github.msemys.esjc.EventStoreTcp.lambda$null$12(EventStoreTcp.java:870) ~[esjc-2.3.0.jar:2.3.0]
	at io.netty.util.concurrent.DefaultPromise.notifyListener0(DefaultPromise.java:511) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListenersNow(DefaultPromise.java:485) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.util.concurrent.DefaultPromise.notifyListeners(DefaultPromise.java:424) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.util.concurrent.DefaultPromise.trySuccess(DefaultPromise.java:103) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.DefaultChannelPromise.trySuccess(DefaultChannelPromise.java:84) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannel$CloseFuture.setClosed(AbstractChannel.java:1152) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannel$AbstractUnsafe.doClose0(AbstractChannel.java:768) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:744) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannel$AbstractUnsafe.close(AbstractChannel.java:615) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.DefaultChannelPipeline$HeadContext.close(DefaultChannelPipeline.java:1350) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:624) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:608) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.ChannelOutboundHandlerAdapter.close(ChannelOutboundHandlerAdapter.java:71) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:624) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:608) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.ChannelOutboundHandlerAdapter.close(ChannelOutboundHandlerAdapter.java:71) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:624) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:608) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.ChannelDuplexHandler.close(ChannelDuplexHandler.java:73) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannelHandlerContext.invokeClose(AbstractChannelHandlerContext.java:624) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:608) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.AbstractChannelHandlerContext.close(AbstractChannelHandlerContext.java:465) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at com.github.msemys.esjc.tcp.handler.HeartbeatHandler.lambda$userEventTriggered$0(HeartbeatHandler.java:61) ~[esjc-2.3.0.jar:2.3.0]
	at io.netty.util.concurrent.PromiseTask$RunnableAdapter.call(PromiseTask.java:38) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.util.concurrent.ScheduledFutureTask.run(ScheduledFutureTask.java:127) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute$$$capture(AbstractEventExecutor.java:163) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.util.concurrent.AbstractEventExecutor.safeExecute(AbstractEventExecutor.java) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:404) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:495) ~[netty-transport-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:905) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) ~[netty-common-4.1.33.Final.jar:4.1.33.Final]
	at java.base/java.lang.Thread.run(Thread.java:832) ~[na:na]

[msemys]

hi, looks like tcp connection was dropped due es server was not responded to ping, that esjc client was sent. if es server is on heavy load it is possible that responses from server side lags.

try to increase esjc heartbeat interval/timeout e.g.:

EventStoreBuilder.newBuilder()
    .heartbeatInterval(Duration.ofSeconds(2))
    .heartbeatTimeout(Duration.ofSeconds(5))
    .build();

[jselamy]

Ok thanks,

Since there's an automatic reconnection set to -1 the client reconnect to the ES.

Perhaps in this same app there's connections to Persistent Subscription and it's not re-subscribing to Persistent Subscriptions. We've added a little mechanism by putting a listener to EventStore instance

es.addListener(event -> {
            if(event instanceof ClientConnected) {
                if (log.isDebugEnabled()) {
                    log.debug("Received EventStore Event: {}", event.getClass().getSimpleName());
                }

                if (log.isInfoEnabled()) {
                    log.info("ClientConnected event received from TCP client sending a ApplicationEvent to notify");
                }

                publisher.publishEvent(new ESClientConnectedEvent(this));
            }
        })

@EventListener
    public void onApplicationEvent(ESClientConnectedEvent event) {
        if(event.isInitialized()) {
            final Map<String, PersistentSubscriber> subscribers = ctx.getBeansOfType(PersistentSubscriber.class);

            subscribers
                    .values()
                    .stream()
                    .peek(ps -> {
                        if(log.isInfoEnabled()) {
                            log.info("Performing a subscribe for stream {} with subscription_name {}", ps.getStream(), ps.getGroupName());
                        }
                    })
                    .forEach(PersistentSubscriber::subscribe);
        }
    }

Would there be a better way to do such a thing ?