SocketAcceptorInterceptor receive an unexpected payload object rather than receive a setup payload object

[KaimingWan]

We create a client and server with reconnection feature. The server register a interceptor to receive the setup payload and then deserialized for authentication.

Client have another daemon thread called Thread A and send data by rsocket periodically. The sended payload data we call it payload-A.

After server is stopped and we can see the priodically sended data is failed as what we expected. After the server is started again, client begin to reconnect to the server. We can see the socket acceptor receive the setup payload data, but it is not the setup data we generated by method getAuthInfo() method. The server sidecar acceptor receive an unexpected normal payload data which is very strange.

Related server and client code:

// client 
        requesterMono = rsocketRequesterBuilder.setupRoute(SETUP_ROUTE)
            .setupData(new ObjectMapper().writeValueAsString(getAuthInfo()))
            .dataMimeType(MimeTypeUtils.parseMimeType(WellKnownMimeType.APPLICATION_JSON.getString()))
            .rsocketConnector(connector -> connector.acceptor(responder).fragment(MTU_BYTE_SIZE)
                .keepAlive(Duration.ofSeconds(KEEP_ALIVE_INTERVAL_SEC), Duration.ofSeconds(KEEP_ALIVE_MAX_TIME_SEC))
                .reconnect(Retry.fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(RSOCKET_RETRY_INTERVAL_SECONDS))
                    .doAfterRetry(retrySignal -> log.warn("RSocket client is reconnecting to get the newest connection...."))))
            .connect(TcpClientTransport.create(TcpClient.create()
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, TCP_CONN_TIMEOUT)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .host(consoleDomain)
                .port(SERVER_PORT)
                .secure(ssl -> ssl.sslContext(sslContext))));

// register a socket acceptor interceptor to receive the authentication setup payload object in the server side
    protected SocketAcceptorInterceptor genAcceptorAuthInterceptor() {
        SocketAcceptorInterceptor interceptor = acceptor -> (setup, sendingSocket) -> {
            try {
                log.info("receive an task connection try to init.");
                LocalTaskAuthInfo authInfo = new ObjectMapper().readValue(setup.getDataUtf8(), LocalTaskAuthInfo.class);

              //ignore some non-related code...
                return acceptor.accept(setup, sendingSocket);
            } catch (Exception e) {
                String errMsg = "try to accept a connection error.msg:" + ExceptionUtils.getRootCauseMessage(e);
                log.warn(errMsg, e);
                exceptionService.saveException(new SidecarRuntimeException(e));
                return Mono.error(new RSocketAuthException(errMsg));
            }
        };
        return interceptor;
    }



//server
 RSocketServer.create(routeHandler.responder()).fragment(RSocketConnConstans.MTU_BYTE_SIZE)
            .interceptors(registry -> registry.forSocketAcceptor(interceptor))
            .bind(TcpServerTransport.create(TcpServer.create()
                .option(EpollChannelOption.SO_KEEPALIVE, true)
                .option(EpollChannelOption.TCP_NODELAY, true)
                .secure(ssl -> ssl.sslContext(sslContext))
                .port(Integer.valueOf(consoleConfig.getRsocketConsolePort()))
                .doOnConnection(
                    t -> log.info("New sidecar CONNECTED. Pay attention that this is maybe a resumed connection..."))))
            .subscribe();

Expected Behavior

Receive setup payload in the socket acceptor

Actual Behavior

Receive a normal payload data.

Steps to Reproduce

Unfortuncately that we can't reproduce it stablely. The issue occur sometimes after server is restarted. Usually after the server is restarted everything goes well.

Possible Solution

May be this is a parallel bug. It occur now and then. I know this issue may ba a little difficult. Is there any suggestion that I can take to set the breakpoint. This issue's key point is to find out why the normal paryload data replace the setup payload data and then cause the deserialization problem.

Your Environment

• RSocket version(s) used: rsocket-core-1.0.2 and springboot-starter-rsocket 2.3.4.RELEASE
• Platform (eg. JVM version (javar -version) or Node version (node --version)): OpenJDK 1.8
• OS and version (eg uname -a): Mac OS 10.15.4

[OlegDokuka]

@KaimingWan it looks like you are observing an already fixed bug (please see #842).

Can you please check if the same is reproducing with 1.0.3 or 1.1.0?

[KaimingWan]

@OlegDokuka Thanks, I will try to upgrade our rsocket version to check whether this issue will happen again. However, I still have some confusion. I checked the issue #842 and find its problem is the cnf count is 0 when reconnecting. His case seems can be reproduced always. My case have some difference because the issue happen now and then. Can you tell me the bug's mechanism and in what situation it will happen? I want to ensure that my issue is caused by the related bug.

[KaimingWan]

@OlegDokuka I updateed my issue. I checked that my rsocket-core version is 1.0.2 and the mentioned bug is fixed. So I think my probleam is not caused by mentioned bug in #842 . Can you give any idea that where I can set the breakpoint to capture the moment when setupPayload is not an object instance of ConnectionSetupPayload. I can set the breakpoint to check why during reconnection my server will receive a non-setup payload contents.

[KaimingWan]

@OlegDokuka To get more information. Would you mind telling me where to set the breakpoint when reconnection haapen and the setup payload is set? I now set the breakpoint in RSocketConnector class's method setupPayload and I find I can't capture any case that I receive a ConnectionSetupPayload.

[KaimingWan]

@OlegDokuka Now I set breakpoint in RSocketServer's acceptSetup method. I will give more info if I encounter this issue again.

[OlegDokuka]

@KaimingWan is it still an issue for you? If so, can you try to reproduce that issue with a pure rsocket setup (I mean your example uses spring-messaging rsocket, however, to minimize the possible error surface, I need to make sure that you observe the same problem with vanilla rsocet-java)

[KaimingWan]

@OlegDokuka Sorry for the late response. The issue happens now and then. Recently I meet the issue again and I put forward the issue here: spring-projects/spring-framework#26626 . May be a little more detailed than this issue I created before. I will try to write an example to show what is going on. But the issue may not happen every time, the example may not reproduce what I met. However, another issue always happen is that the first stream frame log that carry setup payload info is missing when client connect to server. As I expected, the first stream should carry the setup payload and been sent to SETUP_ROUTE( route name is "conn-setup") in my case, but I didn't see any logs. Later I will give an example here.

[KaimingWan]

@OlegDokuka I write a simple example but it seems everything goes well. The first frame carry the setup payload is print out normally. Where to set the break point if I want to debug my real world code. BTW, my test example is as follow:

package com.clougence.test.rsocket;

import static com.clougence.clouddm.comm.constants.rsocket.RSocketConnConstants.MTU_BYTE_SIZE;

import java.time.Duration;

import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.http.codec.json.Jackson2JsonDecoder;
import org.springframework.http.codec.json.Jackson2JsonEncoder;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.RSocketStrategies;
import org.springframework.messaging.rsocket.annotation.support.RSocketMessageHandler;
import org.springframework.util.MimeTypeUtils;
import org.springframework.web.util.pattern.PathPatternRouteMatcher;
import com.clougence.clouddm.base.metadata.auth.ConnAuthDTO;
import com.clougence.clouddm.comm.constants.rsocket.RSocketConnConstants;
import com.clougence.clouddm.comm.util.HostUtil;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.netty.channel.ChannelOption;
import io.netty.channel.epoll.EpollChannelOption;
import io.rsocket.SocketAcceptor;
import io.rsocket.metadata.WellKnownMimeType;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.transport.netty.server.TcpServerTransport;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.netty.tcp.TcpClient;
import reactor.netty.tcp.TcpServer;
import reactor.test.StepVerifier;
import reactor.util.retry.Retry;

/**
 * @author wanshao create time is 2021/3/2
 **/
@SpringBootTest
@Slf4j
public class TestSetupPayload {

    private static Mono<RSocketRequester> requesterMono;

    private final static int              SERVER_PORT                    = 5678;

    private final static int              TCP_CONN_TIMEOUT               = 3000;
    private final static int              RSOCKET_RETRY_INTERVAL_SECONDS = 5;
    private final static int              KEEP_ALIVE_INTERVAL_SEC        = 5;
    private final static int              KEEP_ALIVE_MAX_TIME_SEC        = 30;

    @SneakyThrows
    @BeforeAll
    public static void setupOnce() {
        RSocketStrategies rSocketStrategies = RSocketStrategies.builder()
            .encoders(encoders -> encoders.add(new Jackson2JsonEncoder()))
            .decoders(decoders -> decoders.add(new Jackson2JsonDecoder()))
            .routeMatcher(new PathPatternRouteMatcher())
            .build();
        RSocketRequester.Builder rsocketRequesterBuilder = RSocketRequester.builder().rsocketStrategies(rSocketStrategies);
        ConnAuthDTO connAuthDTO = ConnAuthDTO.builder()
            //
            .ak("myak")
            .sk("mysk")
            .ip(HostUtil.getHostIp())
            .wsn("wsn")
            .sendAuthTimeMs(System.currentTimeMillis())
            .build();

        SocketAcceptor responder = RSocketMessageHandler.responder(rSocketStrategies, new TestSetupPayload());

        // init server
        io.rsocket.core.RSocketServer.create()
            .interceptors(registry -> registry.forSocketAcceptor(new TestSocketAcceptor()))
            .acceptor(responder)
            .fragment(RSocketConnConstants.MTU_BYTE_SIZE)
            .bind(TcpServerTransport.create(TcpServer.create()
                .option(EpollChannelOption.SO_KEEPALIVE, true)
                .option(EpollChannelOption.TCP_NODELAY, true)
                .port(SERVER_PORT)
                .doOnConnection(t -> log.info("NEW WORKER CONNECTED...."))))
            .subscribe();

        // init client

        requesterMono = rsocketRequesterBuilder.setupRoute("conn-setup")
            .setupData(new ObjectMapper().writeValueAsString(connAuthDTO))
            .dataMimeType(MimeTypeUtils.parseMimeType(WellKnownMimeType.APPLICATION_JSON.getString()))
            .rsocketConnector(connector -> connector.acceptor(responder)
                .fragment(MTU_BYTE_SIZE)
                .keepAlive(Duration.ofSeconds(KEEP_ALIVE_INTERVAL_SEC), Duration.ofSeconds(KEEP_ALIVE_MAX_TIME_SEC))
                .reconnect(Retry.fixedDelay(Integer.MAX_VALUE, Duration.ofSeconds(RSOCKET_RETRY_INTERVAL_SECONDS))
                    .doAfterRetry(retrySignal -> log.warn("RSocket client is reconnecting to get the newest connection...."))))
            .connect(TcpClientTransport.create(TcpClient.create()
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, TCP_CONN_TIMEOUT)
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .host("localhost")
                .port(SERVER_PORT)));

    }

    @Test
    public void testRequestGetsResponse() {
        // Send a request message
        Mono<String> result = requesterMono.block().route("default-handler").data("hello").retrieveMono(String.class);

        // Verify that the response message contains the expected data
        StepVerifier.create(result).expectNext("success").verifyComplete();
    }

    @MessageMapping("default-handler")
    Mono<String> requestResponse(final String msg) {
        log.info("Received request-response request: {}", msg);
        return Mono.just("success");
    }

}

test acceptor is :

package com.clougence.test.rsocket;

import org.apache.commons.lang3.exception.ExceptionUtils;
import com.clougence.clouddm.base.metadata.auth.ConnAuthDTO;
import com.fasterxml.jackson.databind.ObjectMapper;
import io.rsocket.SocketAcceptor;
import io.rsocket.plugins.SocketAcceptorInterceptor;
import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;

/**
 * @author wanshao create time is 2021/3/2
 **/
@Slf4j
public class TestSocketAcceptor implements SocketAcceptorInterceptor {

    public TestSocketAcceptor(){
    }

    @Override
    public SocketAcceptor apply(SocketAcceptor socketAcceptor) {
        return (setup, sendingSocket) -> {
            try {
                // Here will occur issue
                ConnAuthDTO authInfo = new ObjectMapper().readValue(setup.getDataUtf8(), ConnAuthDTO.class);

                return socketAcceptor.accept(setup, sendingSocket);
            } catch (Exception e) {
                String errMsg = "try to accept a connection error.msg:" + ExceptionUtils.getRootCauseMessage(e);
                log.warn(errMsg, e);
                return Mono.error(new com.clougence.clouddm.base.metadata.rpc.rsocket.RSocketAuthException(errMsg));
            }
        };
    }

}

The log is ：

[OlegDokuka]

@KaimingWan we had a chat with @rstoyanchev and it can be that this is spring-messaging issue and the fact that the given DTO can be modified (at least that is what I understood from your comments on the linked issue) which means the spring codecs which coverts given object to ByteBuf may provide a corrupted one.

Thus, not sure it is rsocket-java issue