[Feature] k8s环境下还是会出现连接旧pod的ip

[AcceptMediocrity]

Pre-check

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

Search before asking

• I had searched in the issues and found no similar feature requirement.

Apache Dubbo Component

Java SDK (apache/dubbo)

Descriptions

Dubbo：3.2.13
spring-boot: 3.1.5
nacos: 2.2.1
问题：
重启pod的时候还是会出现连接到旧pod的ip

日志：
[Dubbo-framework-connectivity-scheduler-thread-1] ERROR o.a.d.r.t.n.NettyConnectionClient - [?,?] - [DUBBO] Failed to connect to server: /10.194.103.178:18082, dubbo version: 3.2.13, current host: 10.194.98.152, error code: 6-16. This may be caused by , go to https://dubbo.apache.org/faq/6/16 to find instructions. n""}];

Related issues

Dubbo：3.2.13
spring-boot: 3.1.5
nacos: 2.2.1
问题：
重启pod的时候还是会出现连接到旧pod的ip
然后目前的想法是启动一个后台任务主动探活,重新订阅

spi注册RegistryProtocolListener
META-INF/dubbo/internal/org.apache.dubbo.registry.integration.RegistryProtocolListener
k8sRefresh=com.example.K8sConnectivityRefreshListener
代码实现

@activate
public class K8sConnectivityRefreshListener implements RegistryProtocolListener {
private static final Logger logger = LoggerFactory.getLogger(K8sConnectivityRefreshListener.class);

private static final int CHECK_INTERVAL_SECONDS = 10;
private static final int MAX_FAIL_BEFORE_REFRESH = 3; // 连续失败3次才刷新

private final ScheduledExecutorService executor =
        Executors.newSingleThreadScheduledExecutor(
                new NamedThreadFactory("k8s-connectivity-refresh", true));


@Override
public void onExport(RegistryProtocol registryProtocol, Exporter<?> exporter) {

}

@Override
public void onRefer(RegistryProtocol registryProtocol, ClusterInvoker<?> invoker, URL url, URL registryURL) {
    if (!(invoker instanceof MigrationInvoker)) return;

    MigrationInvoker<?> migrationInvoker = (MigrationInvoker<?>) invoker;
    executor.scheduleWithFixedDelay(
            new ConnectivityRefreshTask<>(migrationInvoker, registryProtocol),
            CHECK_INTERVAL_SECONDS,
            CHECK_INTERVAL_SECONDS,
            TimeUnit.SECONDS
    );
}

@Override
public void onDestroy() {
    executor.shutdown();
}

public static class ConnectivityRefreshTask<T> implements Runnable {

    private MigrationInvoker<T> migrationInvoker;

    private RegistryProtocol registryProtocol;

    public ConnectivityRefreshTask(MigrationInvoker<T> migrationInvoker, RegistryProtocol registryProtocol) {
        this.migrationInvoker = migrationInvoker;
        this.registryProtocol = registryProtocol;
    }

    private final Map<String, AtomicInteger> failCountMap = new ConcurrentHashMap<>();

    @Override
    public void run() {
        try {
            ClusterInvoker<T> clusterInvoker = migrationInvoker.getCurrentAvailableInvoker();
            if (!(clusterInvoker.getDirectory() instanceof DynamicDirectory)) return;

            DynamicDirectory<T> directory = (DynamicDirectory<T>) clusterInvoker.getDirectory();

            List<Invoker<T>> allInvokers = directory.getAllInvokers();
            boolean needRefresh = false;

            for (Invoker<T> invoker : allInvokers) {
                String address = invoker.getUrl().getAddress();
                boolean available = invoker.isAvailable();
                logger.info("address:{}, available:{}", address, available);
                if (!available) {
                    int failCount = failCountMap
                            .computeIfAbsent(address, k -> new AtomicInteger())
                            .incrementAndGet();
                    if (failCount >= MAX_FAIL_BEFORE_REFRESH) {
                        needRefresh = true;
                        break;
                    }
                } else {
                    failCountMap.remove(address);
                }
            }
            if (needRefresh) {
                failCountMap.clear();
                Registry registry = directory.getRegistry();
                URL subscribeUrl = directory.getSubscribeUrl();
                registry.subscribe(subscribeUrl, directory);
            }
        } catch (Throwable t) {
            logger.error("K8sConnectivityRefreshListener.ConnectivityRefreshTask.run: ", t);
        }
    }
}

}

Are you willing to submit a pull request to fix on your own?

• Yes I am willing to submit a pull request on my own!

Code of Conduct

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