Ignite瘦客户端集群全节点宕机恢复后无法重连问题咨询

背景 #

我正在评估将Ignite作为我方架构的缓存层，在测试Ignite Java瘦客户端的下述场景时，未在Ignite官方文档及相关社区找到对应处理方案，特此咨询相关参考指引。

适用场景 #

Ignite集群所有节点全部宕机后恢复，即瘦客户端与所有集群节点断开连接一段时间。

预期表现 #

我使用了连续查询（Continuous Query）并注册了断开事件监听，预期会收到断开事件通知，但该事件始终未触发，参考代码如下：

public static QueryCursor<Cache.Entry<String, String>> subscribeForDataUpdates(ClientCache<String, String> entityCache,
                                                                               AtomicLong totalUpdatesTracker) {
    ClientDisconnectListener disconnectListener = reason ->
            System.out.printf("Client: %s received disconnect event with reason:%s %n",
                    getClientIpAddr(),
                    reason.getMessage());
    ContinuousQuery<String, String> continuousQuery = new ContinuousQuery<>();
    continuousQuery.setLocalListener(new CacheUpdateListener(entityCache.getName(), totalUpdatesTracker));
    QueryCursor<Cache.Entry<String, String>> queryCursor = entityCache.query(continuousQuery, disconnectListener);
    System.out.printf("Client: %s - subscribed for change notification(s) for entity cache: %s %n",
            getClientIpAddr(),
            entityCache.getName());
    return queryCursor;
}

现有自定义实现 #

我目前自行实现了连接检测逻辑，定时校验集群连接状态，连接断开后重新初始化瘦客户端并重新订阅连续查询更新，实现代码如下：

import io.vavr.control.Try;
import org.apache.ignite.cache.query.QueryCursor;
import org.apache.ignite.client.IgniteClient;

import javax.cache.Cache;
import javax.inject.Inject;
import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import static com.cisco.ignite.consumer.CacheChangeSubscriber.subscribeForDataUpdates;
import static com.cisco.ignite.consumer.Utils.addShutDownHookToCloseCacheUpdates;
import static com.cisco.ignite.consumer.Utils.getClientIpAddr;

public class ClusterConnectionChecker implements Runnable {
    private static final List<QueryCursor<Cache.Entry<String, String>>> querySubscriptions = new ArrayList<>();
    @Inject
    private CacheChangeSubscriber cacheChangeSubscriber;
    private IgniteClient thinClientInstance;
    private final long secondsDelayBetweenChecks;
    private final List<String> cacheNames;
    private final AtomicLong totalUpdatesTracker;
    private boolean needsReSubscription = false;


    public ClusterConnectionChecker(IgniteClient client, long delayBetweenChecks,
                                    List<String> cacheNames, AtomicLong totalUpdatesTracker) {
        this.thinClientInstance = client;
        this.secondsDelayBetweenChecks = delayBetweenChecks;
        this.cacheNames = cacheNames;
        this.totalUpdatesTracker = totalUpdatesTracker;
    }

    @Override
    public void run() {
        while(!Thread.interrupted()) {
            try {
                Thread.sleep(TimeUnit.SECONDS.toMillis(secondsDelayBetweenChecks));
                boolean isClusterConnectionActive = isConnectionToClusterActive();
                if (!isClusterConnectionActive) {
                    needsReSubscription = true;
                    System.out.printf("Time: %s | Connection to ignite cluster is not active !!! %n",
                            LocalDateTime.now());
                    reInitializeThinClient();
                    reSubscribeForUpdates();
                } else {
                    if (needsReSubscription) {
                        reSubscribeForUpdates();
                    }
                }
            } catch (InterruptedException ie) {
                Thread.currentThread().interrupt();
            }
        }
    }

    private boolean isConnectionToClusterActive() {
        return Try.of(() -> {
            return thinClientInstance.cluster().state().active();
        }).recover(ex -> {
            return false;
        }).getOrElse(false);
    }

    private void reInitializeThinClient() {
        Try.of(() -> {
            thinClientInstance = cacheChangeSubscriber.createThinClientInstance();
            if (thinClientInstance.cluster().state().active()) {
                System.out.printf("Client: %s | Thin client instance was re-initialized since it was not active %n",
                        getClientIpAddr());
            }
            return thinClientInstance;
        }).onFailure(th -> System.out.printf("Client: %s | Failed to re-initialize ignite cluster connection. " +
                "Will re-try after:%d seconds %n", getClientIpAddr(),secondsDelayBetweenChecks));
    }

    private void reSubscribeForUpdates() {
        if (isConnectionToClusterActive()) {
            System.out.printf("Client: %s | Re-subscribing for cache updates after cluster connection re-init... %n",
                    getClientIpAddr());
            totalUpdatesTracker.set(0);
            cacheNames.forEach(name -> querySubscriptions.add(subscribeForDataUpdates(
                    thinClientInstance.getOrCreateCache(name),
                    totalUpdatesTracker)));
            addShutDownHookToCloseCacheUpdates(querySubscriptions, thinClientInstance);
            needsReSubscription = false;
        }
    }
}

内容的提问来源于stack exchange，提问作者Dibakar Sen