<commit_before>package com.github.msemys.esjc.node.cluster;

import com.github.msemys.esjc.node.EndpointDiscoverer;
import com.github.msemys.esjc.node.NodeEndpoints;
import com.github.msemys.esjc.util.Throwables;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonDeserializer;
import com.google.gson.stream.JsonReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.*;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import static com.github.msemys.esjc.util.Preconditions.checkNotNull;
import static java.util.Arrays.asList;
import static java.util.concurrent.TimeUnit.MILLISECONDS;

public class ClusterEndpointDiscoverer implements EndpointDiscoverer {
    private static final Logger logger = LoggerFactory.getLogger(ClusterEndpointDiscoverer.class);

    private final ScheduledExecutorService scheduler;
    private List<MemberInfoDto> oldGossip;
    private final ReentrantLock oldGossipLock = new ReentrantLock();
    private final ClusterNodeSettings settings;
    private final Gson gson;

    public ClusterEndpointDiscoverer(ClusterNodeSettings settings, ScheduledExecutorService scheduler) {
        checkNotNull(settings, "settings is null");
        checkNotNull(scheduler, "scheduler is null");

        this.settings = settings;
        this.scheduler = scheduler;

        gson = new GsonBuilder()
            .registerTypeAdapter(Instant.class,
                (JsonDeserializer<Instant>) (json, type, ctx) -> Instant.parse(json.getAsJsonPrimitive().getAsString()))
            .create();
    }

    @Override
    public CompletableFuture<NodeEndpoints> discover(InetSocketAddress failedTcpEndpoint) {
        CompletableFuture<NodeEndpoints> result = new CompletableFuture<>();

        if (settings.maxDiscoverAttempts != 0) {
            scheduler.execute(() -> discover(result, failedTcpEndpoint, 1));
        } else {
            result.completeExceptionally(new ClusterException("Cluster endpoint discover is not enabled."));
        }

        return result;
    }

    private void discover(CompletableFuture<NodeEndpoints> result, InetSocketAddress failedEndpoint, int attempt) {
        final String attemptInfo = (settings.maxDiscoverAttempts != -1) ?
            String.format("%d/%d", attempt, settings.maxDiscoverAttempts) : String.valueOf(attempt);

        try {
            Optional<NodeEndpoints> nodeEndpoints = tryDiscover(failedEndpoint);

            if (nodeEndpoints.isPresent()) {
                logger.info("Discovering attempt {} successful: best candidate is {}.", attemptInfo, nodeEndpoints.get());
                result.complete(nodeEndpoints.get());
            } else {
                logger.info("Discovering attempt {} failed: no candidate found.", attemptInfo);
            }
        } catch (Exception e) {
            logger.info("Discovering attempt {} failed.", attemptInfo, e);
        }

        if (!result.isDone() && (attempt < settings.maxDiscoverAttempts || settings.maxDiscoverAttempts == -1)) {
            scheduler.schedule(() -> discover(result, failedEndpoint, attempt + 1), settings.discoverAttemptInterval.toMillis(), MILLISECONDS);
        } else {
            result.completeExceptionally(new ClusterException(String.format("Failed to discover candidate in %d attempts.", attempt)));
        }
    }

    private Optional<NodeEndpoints> tryDiscover(InetSocketAddress failedEndpoint) {
        oldGossipLock.lock();
        List<MemberInfoDto> oldGossipCopy;
        try {
            oldGossipCopy = oldGossip;
            oldGossip = null;
        } finally {
            oldGossipLock.unlock();
        }

        List<GossipSeed> gossipCandidates = (oldGossipCopy != null) ?
            getGossipCandidatesFromOldGossip(oldGossipCopy, failedEndpoint) : getGossipCandidatesFromDns();

        Iterator<GossipSeed> iterator = gossipCandidates.iterator();
        while (iterator.hasNext()) {
            Optional<ClusterInfoDto> gossip = tryGetGossipFrom(iterator.next())
                .filter(c -> c.members != null && !c.members.isEmpty());

            if (gossip.isPresent()) {
                Optional<NodeEndpoints> bestNode = tryDetermineBestNode(gossip.get().members);

                if (bestNode.isPresent()) {<commit_msg>maintenance: code cleanup - 'while' loop replaced with 'for-each'<commit_after>package com.github.msemys.esjc.node.cluster;

import com.github.msemys.esjc.node.EndpointDiscoverer;
import com.github.msemys.esjc.node.NodeEndpoints;
import com.github.msemys.esjc.util.Throwables;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonDeserializer;
import com.google.gson.stream.JsonReader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.net.*;
import java.time.Instant;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import static com.github.msemys.esjc.util.Preconditions.checkNotNull;
import static java.util.Arrays.asList;
import static java.util.concurrent.TimeUnit.MILLISECONDS;

public class ClusterEndpointDiscoverer implements EndpointDiscoverer {
    private static final Logger logger = LoggerFactory.getLogger(ClusterEndpointDiscoverer.class);

    private final ScheduledExecutorService scheduler;
    private List<MemberInfoDto> oldGossip;
    private final ReentrantLock oldGossipLock = new ReentrantLock();
    private final ClusterNodeSettings settings;
    private final Gson gson;

    public ClusterEndpointDiscoverer(ClusterNodeSettings settings, ScheduledExecutorService scheduler) {
        checkNotNull(settings, "settings is null");
        checkNotNull(scheduler, "scheduler is null");

        this.settings = settings;
        this.scheduler = scheduler;

        gson = new GsonBuilder()
            .registerTypeAdapter(Instant.class,
                (JsonDeserializer<Instant>) (json, type, ctx) -> Instant.parse(json.getAsJsonPrimitive().getAsString()))
            .create();
    }

    @Override
    public CompletableFuture<NodeEndpoints> discover(InetSocketAddress failedTcpEndpoint) {
        CompletableFuture<NodeEndpoints> result = new CompletableFuture<>();

        if (settings.maxDiscoverAttempts != 0) {
            scheduler.execute(() -> discover(result, failedTcpEndpoint, 1));
        } else {
            result.completeExceptionally(new ClusterException("Cluster endpoint discover is not enabled."));
        }

        return result;
    }

    private void discover(CompletableFuture<NodeEndpoints> result, InetSocketAddress failedEndpoint, int attempt) {
        final String attemptInfo = (settings.maxDiscoverAttempts != -1) ?
            String.format("%d/%d", attempt, settings.maxDiscoverAttempts) : String.valueOf(attempt);

        try {
            Optional<NodeEndpoints> nodeEndpoints = tryDiscover(failedEndpoint);

            if (nodeEndpoints.isPresent()) {
                logger.info("Discovering attempt {} successful: best candidate is {}.", attemptInfo, nodeEndpoints.get());
                result.complete(nodeEndpoints.get());
            } else {
                logger.info("Discovering attempt {} failed: no candidate found.", attemptInfo);
            }
        } catch (Exception e) {
            logger.info("Discovering attempt {} failed.", attemptInfo, e);
        }

        if (!result.isDone() && (attempt < settings.maxDiscoverAttempts || settings.maxDiscoverAttempts == -1)) {
            scheduler.schedule(() -> discover(result, failedEndpoint, attempt + 1), settings.discoverAttemptInterval.toMillis(), MILLISECONDS);
        } else {
            result.completeExceptionally(new ClusterException(String.format("Failed to discover candidate in %d attempts.", attempt)));
        }
    }

    private Optional<NodeEndpoints> tryDiscover(InetSocketAddress failedEndpoint) {
        oldGossipLock.lock();
        List<MemberInfoDto> oldGossipCopy;
        try {
            oldGossipCopy = oldGossip;
            oldGossip = null;
        } finally {
            oldGossipLock.unlock();
        }

        List<GossipSeed> gossipCandidates = (oldGossipCopy != null) ?
            getGossipCandidatesFromOldGossip(oldGossipCopy, failedEndpoint) : getGossipCandidatesFromDns();

        for (GossipSeed gossipCandidate : gossipCandidates) {
            Optional<ClusterInfoDto> gossip = tryGetGossipFrom(gossipCandidate)
                .filter(c -> c.members != null && !c.members.isEmpty());

            if (gossip.isPresent()) {
                Optional<NodeEndpoints> bestNode = tryDetermineBestNode(gossip.get().members);

                if (bestNode.isPresent()) {
{'fraction_non_alphanumeric': 0.10286401399212942, 'fraction_numerical': 0.001530389156099694, 'mean_word_length': 3.3670644391408113, 'pattern_counts': {'":': 0, '<': 30, '<?xml version=': 0, '>': 36, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4743765', 'n_tokens_mistral': 2566, 'n_tokens_neox': 2465, 'n_words': 496}