Reconnection of a callback function on the consumer after a provider disconnect-reconnect

[mitpjones]

Pre-check

• I am sure that all the content I provide is in English.

Apache Dubbo Component

Java SDK (apache/dubbo)

Details

Hi,

when a Provider disconnects, a service annotated with @DubboReference behaves as expected i.e. if the Consumer calls a method on the proxy a RpcException is thrown. When the Provider reconnects, methods on the service can be called again with no issue.

However, if it is a Callback service (ref https://dubbo.apache.org/en/overview/mannual/java-sdk/tasks/framework/more/callback-parameter/), after reconnecting, the listener will not be invoked when the Callback service is invoked by the Provider even though I can see from the log on the consumer a "Received invokers changed event from registry Urls Size : 1. Invokers Size : 1. Available Size: 1 ..."

Is this a bug? Or is this expected behaviour? If so, what is the recommended way of detecting a reconnection event on the Consumer and then 'reconnecting' the listener to the proxy (or other)?

Regards,

Tim

Code of Conduct

• I agree to follow this project's Code of Conduct

[zrlw]

do you test like this?

@DubboReference
private CallbackService service;

@TestBoost public void test() throws InterruptedException {
    while (true) {
        CountDownLatch latch = new CountDownLatch(1);
        List<String> results = new ArrayList<>();
        service.addListener("foo.bar", msg -> {
            results.add(msg);
            latch.countDown();
        });

        latch.await(5000, TimeUnit.MILLISECONDS);
        Assert.assertFalse(results.isEmpty());
        Assert.assertTrue(results.get(0).startsWith("Changed:"));
        
        // wait for provider reconnection?
        Thread.sleep(10000);
    }
}

[mitpjones]

The test above also does not work and results in the following exception after the provider disconnects and then terminates

Exception in thread "main" org.apache.dubbo.rpc.RpcException: No provider available from registry RegistryDirectory(registry: 127.0.0.1:2181)-Directory(invokers: 0[], validInvokers: 0[], invokersToReconnect: 0[]) for service org.apache.dubbo.samples.callback.api.CallbackService on consumer 172.28.5.128 use dubbo version 3.3.1, please check status of providers(disabled, not registered or in blocklist). at org.apache.dubbo.registry.integration.DynamicDirectory.doList(DynamicDirectory.java:204) at org.apache.dubbo.rpc.cluster.directory.AbstractDirectory.list(AbstractDirectory.java:232) ... at org.apache.dubbo.config.spring.util.LazyTargetInvocationHandler.invoke(LazyTargetInvocationHandler.java:54) at org.apache.dubbo.samples.callback.api.CallbackServiceDubboProxy0.addListener(CallbackServiceDubboProxy0.java) at org.apache.dubbo.samples.callback.consumer.Task.run(Task.java:47) at org.springframework.boot.SpringApplication.lambda$callRunner$5(SpringApplication.java:790) ... at org.springframework.boot.SpringApplication.run(SpringApplication.java:1354) at org.springframework.boot.SpringApplication.run(SpringApplication.java:1343) at org.apache.dubbo.samples.callback.consumer.ConsumerApplication.main(ConsumerApplication.java:31)

Although the test above can be modified to work (surround service.addListener with catch RpcException) I do not think it would be a very good solution. I don't think the application code should be expected to implement a loop with a sleep as it is 'leaking' reconnection concerns into the application code. I would also be concerned with a potential memory leak if this was to be the suggested solution, I did a quick check using Yourkit memory profiler and could see the object count increasing each time service.addListener() was called.

If we think about the standard (non callback) consumer DubboReference it will cope with reconnections without any extra code implemented by the application. I think the same should happen for a callback i.e. the Dubbo library code should handle the reconnection (this may entail it being responsible for it re adding the listener or other?)

In the mean time, the best I could think of was to periodically use the Echo service to check for connectivity, and re add the listener if necessary as follows

@Component
public class VersionChangeCallbackManager {    

    private final ScheduledExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
    private static final long RETRY_DELAY_SECONDS = 1;
    private Future<?> callbackMonitorFuture;

    public void setUpCallbackListener(final VersionChangeCallbackService versionChangeCallbackService) {

        // Needs to be unique across all consumers
        final String listenerKey = "apache.dubbo.consumer.version.change." + UUID.randomUUID();

        final VersionChange versionChange = new VersionChange() {

            @Override
            public void versionChanged(ServerStatus serverStatus) {
                System.out.println("ServerStatus: " + serverStatus);
            }

        };

        callbackMonitorFuture = executorService.scheduleWithFixedDelay(new Runnable() {

            // set flag to true so that the first time through the loop the listener will be added
            private boolean isDisconnected = true;

            @Override
            public void run() {
                try {
                    try {
                        // Attempt to echo to verify connection
                        EchoService echoService = (EchoService) versionChangeCallbackService;
                        String status = (String) echoService.$echo("OK");
                        //System.out.println("setUpCallbackListener() - CallbackService status: " + status);

                        //System.out.println("setUpCallbackListener() - isDisconnected: " + isDisconnected);

                        if (isDisconnected) {
                            System.out.println("setUpCallbackListener() - Reconnected: adding listener with listenerKey: " + listenerKey);
                            versionChangeCallbackService.addListener(listenerKey,
                                                                     versionChange);
                            isDisconnected = false;
                        }

                    }
                    // An RpcException will be thrown when invoking echoService.$echo() if the Provider has disconnected
                    catch (RpcException rpcException) {
                        if (!isDisconnected) {
                            System.out.println("setUpCallbackListener() - Lost connection to CallbackService: " + rpcException.getMessage());
                            rpcException.printStackTrace();
                        }
                        isDisconnected = true;
                    }

                } catch (Exception e) {
                    System.out.println("setUpCallbackListener() - Unexpected error in callback monitoring task");
                    e.printStackTrace();
                }
            }
        }, 0, RETRY_DELAY_SECONDS, TimeUnit.SECONDS);
    }

    public void shutdown() {
        if (callbackMonitorFuture != null) {
            callbackMonitorFuture.cancel(true);
        }
        executorService.shutdown();
        try {
            if (!executorService.awaitTermination(5, TimeUnit.SECONDS)) {
                executorService.shutdownNow();
            }
        } catch (InterruptedException e) {
            executorService.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}

Usage

    @DubboReference
    VersionChangeCallbackService versionChangeCallbackService;

    @Autowired
    VersionChangeCallbackManager versionChangeCallbackManager;

    @Override
    public void run(String... args) throws Exception {

        versionChangeCallbackManager.setUpCallbackListener(versionChangeCallbackService);
        

However I think the Dubbo library could do better than this as by the looks of the logging I think it must know when a disconnection/reconnection event occurs and hence be able to hook into that somehow.

[zrlw]

As for me, it's not dubbo client duty to reconnect because callback is triggered by the server side, let say, kill the service and restart the service, how could it know there are clients are waiting for the callback?

[zrlw]

If we think about the standard (non callback) consumer DubboReference it will cope with reconnections without any extra code implemented by the application.

you are wrong, if the connection is broken, RpcException with No provider available will also be thrown immediately when you using standard DubboReference (non callback), it doesn't try reconnection when the connection is broken.

[mitpjones]

Okay fair point, but you can catch the RpcException and keep retrying until the Provider is back up, no extra application level 'reconnection' code is required.

For example if you modify the code in the non Callback example https://github.com/apache/dubbo-samples/blob/master/1-basic/dubbo-samples-spring-boot/dubbo-samples-spring-boot-consumer/src/main/java/org/apache/dubbo/springboot/demo/consumer/Task.java to catch the RpcException upon a disconnect as follows

            new Thread(()-> {
                while (true) {
                    try {
                        Thread.sleep(1000);
                        try {
                            System.out.println(new Date() + " Receive result ======> " + demoService.sayHello("world"));
                        }
                        catch (RpcException rpcException) {
                            System.out.println("Unexpected exception when trying to connect to server. " + rpcException.getMessage());
                            rpcException.printStackTrace();
                            Thread.sleep(1000);
                        }
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                        Thread.currentThread().interrupt();
                    }
                }
            }).start();

you can then stop the Provider, thus causing a disconnect and then restart the Provider and the Consumer will happily continue. No extra application code is required to handle the disconnection.

For a Callback you have to stop the Consumer and restart it which I think is a poor solution, or implement something like I have shown earlier. Is there a better solution? Perhaps an example would be good as I am sure handling disconnections would be a common requirement among other Dubbo users.

My guess (rightly or wrongly) is that the Callback is implemented using a long running connection. If that is so, I was hoping for something akin to the automatic reconnect functionality of Server Sent Events https://html.spec.whatwg.org/multipage/server-sent-events.html where the EventSource API provides the capability of Automatic reconnection i.e If the connection is lost, the EventSource will automatically attempt to reconnect with an exponential backoff strategy.

[zrlw]

ReconnectTimerTask does what you said, but it has nothing to do with current rpc operation.