@GuardedBy("isolatedEndpointsSetLock")
private final Set<Endpoints> isolatedEndpointsSet;
private final ReadWriteLock isolatedEndpointsSetLock;

public ProducerImpl(String group) throws ClientException {
    super(group);
    this.defaultSendCallbackExecutor = new ThreadPoolExecutor(
            Runtime.getRuntime().availableProcessors(),
            Runtime.getRuntime().availableProcessors(),
            60,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<Runnable>(),
            new ThreadFactoryImpl("SendCallbackWorker"));

    this.sendingRouteDataCache = new ConcurrentHashMap<String, SendingTopicRouteData>();

    this.isolatedEndpointsSet = new HashSet<Endpoints>();
    this.isolatedEndpointsSetLock = new ReentrantReadWriteLock();
}

/**
* Check the status of isolated {@link Endpoints}, rejoin it if it is healthy.
*/
@Override
public void doHealthCheck() {
    final Set<Endpoints> routeEndpointsSet = getRouteEndpointsSet();
    final Set<Endpoints> expired = new HashSet<Endpoints>(Sets.difference(routeEndpointsSet, isolatedEndpointsSet));
    // remove all isolated endpoints which is expired.
    isolatedEndpointsSetLock.writeLock().lock();
    try {
        isolatedEndpointsSet.removeAll(expired);
    } finally {
        isolatedEndpointsSetLock.writeLock().unlock();
    }

    HealthCheckRequest request = HealthCheckRequest.newBuilder().build();
    isolatedEndpointsSetLock.readLock().lock();
    try {
        for (final Endpoints endpoints : isolatedEndpointsSet) {
            Metadata metadata;
            try {
                metadata = sign();
            } catch (Throwable t) {
                continue;
            }
            final ListenableFuture<HealthCheckResponse> future =
                    clientManager.healthCheck(endpoints, metadata, request, ioTimeoutMillis, TimeUnit.MILLISECONDS);
            Futures.addCallback(future, new FutureCallback<HealthCheckResponse>() {
                @Override
                public void onSuccess(HealthCheckResponse response) {
                    final Status status = response.getCommon().getStatus();
                    final Code code = Code.forNumber(status.getCode());
                    // target endpoints is healthy, rejoin it.
                    if (Code.OK.equals(code)) {
                        isolatedEndpointsSetLock.writeLock().lock();
                        try {
                            isolatedEndpointsSet.remove(endpoints);
                        } finally {
                            isolatedEndpointsSetLock.writeLock().unlock();
                        }
                        log.info("Rejoin endpoints which is isolated before, clientId={}, endpoints={}", id,
                                endpoints);
                        return;
                    }
                    log.warn("Failed to rejoin the endpoints which is isolated before, clientId={}, code={}, "
                            + "status message=[{}], endpoints={}", id, code, status.getMessage(), endpoints);
                }

                @Override
                public void onFailure(Throwable t) {
                    log.error("Failed to do health check, clientId={}, endpoints={}", id, endpoints, t);
                }
            });
        }
    } finally {
        isolatedEndpointsSetLock.readLock().unlock();
    }
}

package me.tianshuang;

public class ApiResult<T> {
    private T data;

    public T getData() {
        return data;
    }

    public void setData(T data) {
        this.data = data;
    }

    @Override
    public String toString() {
        return "ApiResult{" +
                "data=" + data +
                '}';
    }
}