src/test/java/io/axoniq/axonserver/connector/impl/ControlChannelIntegrationTest.java
/*
      Verifies that a connection is force-reset when an UNAVAILABLE error is returned on the Control Channel,
      while the connection itself reports to be READY. This may indicate a proxy is keeping the connection open,
      but is rejecting requests due to an unavailable backend.
     */
Verifies that a connection is force-reset when an UNAVAILABLE error is returned on the Control Channel,
while the connection itself reports to be READY. This may indicate a proxy is keeping the connection open,
but is rejecting requests due to an unavailable backend.
Verifies that a connection is force - reset when an UNAVAILABLE error is returned on the Control Channel while the connection itself reports to be READY . This may indicate a proxy is keeping the connection open but is rejecting requests due to an unavailable backend .
@Test
    void connectionForcefullyRecreatedAfterFailureOnInstructionChannelAndLiveChannel() throws InterruptedException {
        AtomicInteger connectCounter = new AtomicInteger();
        CallCancellingInterceptor cancellingInterceptor = new CallCancellingInterceptor();
        client = AxonServerConnectionFactory.forClient("handler")
                                            .routingServers(axonServerAddress)
                                            .connectTimeout(1500, TimeUnit.MILLISECONDS)
                                            .processorInfoUpdateFrequency(500, TimeUnit.MILLISECONDS)
                                            .reconnectInterval(50, TimeUnit.MILLISECONDS)
                                            .customize(mcb -> {
                                                synchronized (this) {
                                                    connectCounter.incrementAndGet();
                                                           return mcb.intercept(cancellingInterceptor);
                                                       }
                                                   })
                                                   .build();
        AxonServerConnection connection1 = client.connect("default");
        assertEquals(1, connectCounter.get());
        QueryChannelImpl handlerClientQueryChannel = (QueryChannelImpl) connection1.queryChannel();
        CommandChannelImpl handlerClientCommandChannel = (CommandChannelImpl) connection1.commandChannel();
        ControlChannelImpl controlChannel = (ControlChannelImpl) connection1.controlChannel();
        handlerClientQueryChannel.registerQueryHandler((query, responseHandler) -> responseHandler.sendLast(QueryResponse.newBuilder().setPayload(SerializedObject.newBuilder().setData(ByteString.copyFromUtf8("Response"))).build()), new QueryDefinition("echo", String.class));
        handlerClientCommandChannel.registerCommandHandler(c -> CompletableFuture.completedFuture(CommandResponse.newBuilder().setPayload(SerializedObject.newBuilder().setData(ByteString.copyFromUtf8("Response"))).build()), 100, "echo");
        assertWithin(2, TimeUnit.SECONDS, () -> assertTrue(connection1.isReady()));
        logger.info("Connection status is READY");
        assertEquals(1, connectCounter.get());
        for (int i = 0; i < 10; i++) {
            assertWithin(2, TimeUnit.SECONDS, () -> assertTrue(connection1.isReady()));
            assertWithin(1, TimeUnit.SECONDS, () -> assertTrue(controlChannel.isReady()));
            logger.info("Simulating failing calls on successful connection");
            cancellingInterceptor.cancelAll(Status.UNAVAILABLE);
        }
        logger.info("Waiting for connector to establish connection");
        assertWithin(1, TimeUnit.SECONDS, () -> assertTrue(controlChannel.isReady()));
        assertWithin(1, TimeUnit.SECONDS, () -> assertTrue(handlerClientQueryChannel.isReady()));
        assertWithin(1, TimeUnit.SECONDS, () -> assertTrue(handlerClientCommandChannel.isReady()));
        assertEquals(11, connectCounter.get());
    }
@ Test void connectionForcefullyRecreatedAfterFailureOnInstructionChannelAndLiveChannel ( ) throws InterruptedException { AtomicInteger connectCounter = new AtomicInteger ( ) ; CallCancellingInterceptor cancellingInterceptor = new CallCancellingInterceptor ( ) ; client = AxonServerConnectionFactory . forClient ( "handler" ) . routingServers ( axonServerAddress ) . connectTimeout ( 1500 , TimeUnit . MILLISECONDS ) . processorInfoUpdateFrequency ( 500 , TimeUnit . MILLISECONDS ) . reconnectInterval ( 50 , TimeUnit . MILLISECONDS ) . customize ( mcb -> { synchronized ( this ) { connectCounter . incrementAndGet ( ) ; return mcb . intercept ( cancellingInterceptor ) ; } } ) . build ( ) ; AxonServerConnection connection1 = client . connect ( "default" ) ; assertEquals ( 1 , connectCounter . get ( ) ) ; QueryChannelImpl handlerClientQueryChannel = ( QueryChannelImpl ) connection1 . queryChannel ( ) ; CommandChannelImpl handlerClientCommandChannel = ( CommandChannelImpl ) connection1 . commandChannel ( ) ; ControlChannelImpl controlChannel = ( ControlChannelImpl ) connection1 . controlChannel ( ) ; handlerClientQueryChannel . registerQueryHandler ( ( query , responseHandler ) -> responseHandler . sendLast ( QueryResponse . newBuilder ( ) . setPayload ( SerializedObject . newBuilder ( ) . setData ( ByteString . copyFromUtf8 ( "Response" ) ) ) . build ( ) ) , new QueryDefinition ( "echo" , String . class ) ) ; handlerClientCommandChannel . registerCommandHandler ( c -> CompletableFuture . completedFuture ( CommandResponse . newBuilder ( ) . setPayload ( SerializedObject . newBuilder ( ) . setData ( ByteString . copyFromUtf8 ( "Response" ) ) ) . build ( ) ) , 100 , "echo" ) ; assertWithin ( 2 , TimeUnit . SECONDS , ( ) -> assertTrue ( connection1 . isReady ( ) ) ) ; logger . info ( "Connection status is READY" ) ; assertEquals ( 1 , connectCounter . get ( ) ) ; for ( int i = 0 ; i < 10 ; i ++ ) { assertWithin ( 2 , TimeUnit . SECONDS , ( ) -> assertTrue ( connection1 . isReady ( ) ) ) ; assertWithin ( 1 , TimeUnit . SECONDS , ( ) -> assertTrue ( controlChannel . isReady ( ) ) ) ; logger . info ( "Simulating failing calls on successful connection" ) ; cancellingInterceptor . cancelAll ( Status . UNAVAILABLE ) ; } logger . info ( "Waiting for connector to establish connection" ) ; assertWithin ( 1 , TimeUnit . SECONDS , ( ) -> assertTrue ( controlChannel . isReady ( ) ) ) ; assertWithin ( 1 , TimeUnit . SECONDS , ( ) -> assertTrue ( handlerClientQueryChannel . isReady ( ) ) ) ; assertWithin ( 1 , TimeUnit . SECONDS , ( ) -> assertTrue ( handlerClientCommandChannel . isReady ( ) ) ) ; assertEquals ( 11 , connectCounter . get ( ) ) ; }
Verifies that a connection is force-reset when an UNAVAILABLE error is returned on the Control Channel,
while the connection itself reports to be READY.
Verifies that a connection is force - reset when an UNAVAILABLE error is returned on the Control Channel while the connection itself reports to be READY .
// waiting to recover