<commit_before>/*
 * Copyright Terracotta, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.terracotta.voter;

import com.tc.voter.VoterManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static java.util.Arrays.asList;
import static java.util.stream.Collectors.toCollection;

/**
 * ActiveVoter votes only for the active it originally connects to.  If the active disconnects,
 * vote for the next server requesting vote.  If the server with the new vote becomes the new active,
 * continue voting for that active.  If the server with the new vote does not become active
 * in a reasonable amount of time, disconnect from all server and hunt for the new active until found.
 */
public class ActiveVoter implements AutoCloseable {

  private static final Logger LOGGER = LoggerFactory.getLogger(ActiveVoter.class);

  private final static String ACTIVE_COORDINATOR = "ACTIVE-COORDINATOR";

  private static final long HEARTBEAT_INTERVAL = 1000L;
  private static final long REG_RETRY_INTERVAL = 5000L;
  private static final long DEFAULT_TOPOLOGY_FETCH_TIME = 300000L;
  public static final String TOPOLOGY_FETCH_TIME_PROPERTY = "org.terracotta.voter.topology.fetch.interval";
  private static final long topologyFetchInterval = Long.getLong(TOPOLOGY_FETCH_TIME_PROPERTY, DEFAULT_TOPOLOGY_FETCH_TIME);

  private final Thread voter;
  private final String id;
  private final Function<String, ClientVoterManager> clientVoterManagerFactory;
  private volatile boolean active = false;
  private volatile Future<?> topologyFetchingFuture = null;
  private final Set<String> existingTopology = new HashSet<>();
  private final List<ClientVoterManager> voterManagers = new CopyOnWriteArrayList<>();
  private final Set<String> registrationLatch = new HashSet<>();
  private final Map<String, Future<?>> heartbeatFutures = new ConcurrentHashMap<>();

  public ActiveVoter(String id, CompletableFuture<VoterStatus> voterStatus, Optional<Properties> connectionProps, String... hostPorts) {
    this(id, voterStatus, connectionProps, ClientVoterManagerImpl::new, hostPorts);
  }

  public ActiveVoter(String id, CompletableFuture<VoterStatus> voterStatus, Optional<Properties> connectionProps,
                     Function<String, ClientVoterManager> clientVoterManagerFactory, String... hostPorts) {
    this.voter = voterThread(voterStatus, connectionProps, hostPorts);
    this.id = id;
    this.clientVoterManagerFactory = clientVoterManagerFactory;
  }

  public ActiveVoter start() {
    this.voter.start();
    return this;
  }

  private Thread voterThread(CompletableFuture<VoterStatus> voterStatus, Optional<Properties> connectionProps, String... hostPorts) {
    return new Thread(() -> {
      ExecutorService executorService = Executors.newCachedThreadPool();
      Stream.of(hostPorts).map(clientVoterManagerFactory).collect(toCollection(() -> voterManagers));
      existingTopology.addAll(new HashSet<>(asList(hostPorts)));
      registrationLatch.addAll(new HashSet<>(asList(hostPorts)));
      try {
        while (!Thread.currentThread().isInterrupted()) {
          ClientVoterManager currentActive = registerWithActive(id, voterManagers, connectionProps);
          active = true;
          LOGGER.info("{} registered with the active: {}", this, currentActive.getTargetHostPort());
          voterStatus.complete(new VoterStatus() {
            @Override
            public boolean isActive() {
              return active;
            }

            @Override
            public void awaitRegistrationWithAll() throws InterruptedException {
              synchronized (registrationLatch) {
                while (!registrationLatch.isEmpty()) {
                  registrationLatch.wait();
                }
              }
            }

            @Override
            public void awaitRegistrationWithAll(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
              long current = System.currentTimeMillis();
              long finish = current + unit.toMillis(timeout);
              synchronized (registrationLatch) {
                while (!registrationLatch.isEmpty()) {
                  current = System.currentTimeMillis();
                  if (System.currentTimeMillis() >= finish) {
                    throw new TimeoutException("Registration timed out");
                  }
                  registrationLatch.wait(finish - current);
                }
              }
            }
          });

          registerAndHeartbeat(executorService, currentActive, connectionProps);
          active = false;
        }
      } catch (InterruptedException e) {
        LOGGER.warn("{} interrupted", this);
      }
      active = false;
      cleanHeartBeatingAndPollingFutures();
      executorService.shutdownNow();
      try {
        executorService.awaitTermination(5, TimeUnit.SECONDS);
      } catch (InterruptedException ie) {
        LOGGER.warn("{} interrupted", this);
      }
    });
  }

  ClientVoterManager registerWithActive(String id,
                                        List<ClientVoterManager> voterManagers, Optional<Properties> connectionProps) throws InterruptedException {
    CompletableFuture<ClientVoterManager> registrationLatch = new CompletableFuture<>();
    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(voterManagers.size());

    List<ScheduledFuture<?>> futures = voterManagers.stream().map(voterManager -> executorService.scheduleAtFixedRate(() -> {
      if (!voterManager.isConnected()) {
        voterManager.connect(connectionProps);
      }

      if (!registrationLatch.isDone()) {
        try {
          String serverState = voterManager.getServerState();
          if (serverState.equals(ACTIVE_COORDINATOR)) {
            long response = voterManager.registerVoter(id);
            if (response >= 0) {
              registrationLatch.complete(voterManager);
            } else {
              StringBuilder message = new StringBuilder();
              message.append(String.format("Registration with %s in state %s failed. ",
                  voterManager.getTargetHostPort(), voterManager.getServerState()));
              long voterLimit = Math.max(0, voterManager.getRegisteredVoterLimit());
              if (voterManager.getRegisteredVoterCount() >= voterLimit) {
                message.append(String.format("Configured voter limit (%d) has already been reached. ", voterLimit));
              }
              message.append("Retrying...");
              LOGGER.warn(message.toString());
            }
          } else {
            LOGGER.info("State of {}: {}. Continuing the search for an active server.", voterManager.getTargetHostPort(), serverState);
          }
        } catch (TimeoutException e) {<commit_msg>don't throw on failure to connect during registeration attempts<commit_after>/*
 * Copyright Terracotta, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.terracotta.voter;

import com.tc.voter.VoterManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static java.util.Arrays.asList;
import static java.util.stream.Collectors.toCollection;

/**
 * ActiveVoter votes only for the active it originally connects to.  If the active disconnects,
 * vote for the next server requesting vote.  If the server with the new vote becomes the new active,
 * continue voting for that active.  If the server with the new vote does not become active
 * in a reasonable amount of time, disconnect from all server and hunt for the new active until found.
 */
public class ActiveVoter implements AutoCloseable {

  private static final Logger LOGGER = LoggerFactory.getLogger(ActiveVoter.class);

  private final static String ACTIVE_COORDINATOR = "ACTIVE-COORDINATOR";

  private static final long HEARTBEAT_INTERVAL = 1000L;
  private static final long REG_RETRY_INTERVAL = 5000L;
  private static final long DEFAULT_TOPOLOGY_FETCH_TIME = 300000L;
  public static final String TOPOLOGY_FETCH_TIME_PROPERTY = "org.terracotta.voter.topology.fetch.interval";
  private static final long topologyFetchInterval = Long.getLong(TOPOLOGY_FETCH_TIME_PROPERTY, DEFAULT_TOPOLOGY_FETCH_TIME);

  private final Thread voter;
  private final String id;
  private final Function<String, ClientVoterManager> clientVoterManagerFactory;
  private volatile boolean active = false;
  private volatile Future<?> topologyFetchingFuture = null;
  private final Set<String> existingTopology = new HashSet<>();
  private final List<ClientVoterManager> voterManagers = new CopyOnWriteArrayList<>();
  private final Set<String> registrationLatch = new HashSet<>();
  private final Map<String, Future<?>> heartbeatFutures = new ConcurrentHashMap<>();

  public ActiveVoter(String id, CompletableFuture<VoterStatus> voterStatus, Optional<Properties> connectionProps, String... hostPorts) {
    this(id, voterStatus, connectionProps, ClientVoterManagerImpl::new, hostPorts);
  }

  public ActiveVoter(String id, CompletableFuture<VoterStatus> voterStatus, Optional<Properties> connectionProps,
                     Function<String, ClientVoterManager> clientVoterManagerFactory, String... hostPorts) {
    this.voter = voterThread(voterStatus, connectionProps, hostPorts);
    this.id = id;
    this.clientVoterManagerFactory = clientVoterManagerFactory;
  }

  public ActiveVoter start() {
    this.voter.start();
    return this;
  }

  private Thread voterThread(CompletableFuture<VoterStatus> voterStatus, Optional<Properties> connectionProps, String... hostPorts) {
    return new Thread(() -> {
      ExecutorService executorService = Executors.newCachedThreadPool();
      Stream.of(hostPorts).map(clientVoterManagerFactory).collect(toCollection(() -> voterManagers));
      existingTopology.addAll(new HashSet<>(asList(hostPorts)));
      registrationLatch.addAll(new HashSet<>(asList(hostPorts)));
      try {
        while (!Thread.currentThread().isInterrupted()) {
          ClientVoterManager currentActive = registerWithActive(id, voterManagers, connectionProps);
          active = true;
          LOGGER.info("{} registered with the active: {}", this, currentActive.getTargetHostPort());
          voterStatus.complete(new VoterStatus() {
            @Override
            public boolean isActive() {
              return active;
            }

            @Override
            public void awaitRegistrationWithAll() throws InterruptedException {
              synchronized (registrationLatch) {
                while (!registrationLatch.isEmpty()) {
                  registrationLatch.wait();
                }
              }
            }

            @Override
            public void awaitRegistrationWithAll(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
              long current = System.currentTimeMillis();
              long finish = current + unit.toMillis(timeout);
              synchronized (registrationLatch) {
                while (!registrationLatch.isEmpty()) {
                  current = System.currentTimeMillis();
                  if (System.currentTimeMillis() >= finish) {
                    throw new TimeoutException("Registration timed out");
                  }
                  registrationLatch.wait(finish - current);
                }
              }
            }
          });

          registerAndHeartbeat(executorService, currentActive, connectionProps);
          active = false;
        }
      } catch (InterruptedException e) {
        LOGGER.warn("{} interrupted", this);
      }
      active = false;
      cleanHeartBeatingAndPollingFutures();
      executorService.shutdownNow();
      try {
        executorService.awaitTermination(5, TimeUnit.SECONDS);
      } catch (InterruptedException ie) {
        LOGGER.warn("{} interrupted", this);
      }
    });
  }

  ClientVoterManager registerWithActive(String id,
                                        List<ClientVoterManager> voterManagers, Optional<Properties> connectionProps) throws InterruptedException {
    CompletableFuture<ClientVoterManager> registrationLatch = new CompletableFuture<>();
    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(voterManagers.size());

    List<ScheduledFuture<?>> futures = voterManagers.stream().map(voterManager -> executorService.scheduleAtFixedRate(() -> {
      if (!voterManager.isConnected()) {
        try {
          voterManager.connect(connectionProps);
        } catch (Exception e) {
          LOGGER.warn("unable to connect to server", e);
          return;
        }
      }

      if (!registrationLatch.isDone()) {
        try {
          String serverState = voterManager.getServerState();
          if (serverState.equals(ACTIVE_COORDINATOR)) {
            long response = voterManager.registerVoter(id);
            if (response >= 0) {
              registrationLatch.complete(voterManager);
            } else {
              StringBuilder message = new StringBuilder();
              message.append(String.format("Registration with %s in state %s failed. ",
                  voterManager.getTargetHostPort(), voterManager.getServerState()));
              long voterLimit = Math.max(0, voterManager.getRegisteredVoterLimit());
              if (voterManager.getRegisteredVoterCount() >= voterLimit) {
                message.append(String.format("Configured voter limit (%d) has already been reached. ", voterLimit));
              }
              message.append("Retrying...");
              LOGGER.warn(message.toString());
            }
          } else {
            LOGGER.info("State of {}: {}. Continuing the search for an active server.", voterManager.getTargetHostPort(), serverState);
          }
        } catch (TimeoutException e) {
{'fraction_non_alphanumeric': 0.09087584567562626, 'fraction_numerical': 0.002803681355518986, 'mean_word_length': 3.2518787250583054, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 69, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '674991', 'n_tokens_mistral': 4276, 'n_tokens_neox': 4159, 'n_words': 1095}