Reset known leader when command fails due to ConnectException to ensu…

…re new leaders can be found quickly.

protocols/raft/src/main/java/io/atomix/protocols/raft/RaftMetadataClient.java

@@ -36,11 +36,20 @@ public interface RaftMetadataClient {
   MemberId getLeader();
 
   /**
-   * Returns the set of known servers in the cluster.
+   * Returns the set of known members in the cluster.
    *
-   * @return The set of known servers in the cluster.
+   * @return The set of known members in the cluster.
    */
-  Collection<MemberId> getServers();
+  default Collection<MemberId> getServers() {
+    return getMembers();
+  }
+
+  /**
+   * Returns the set of known members in the cluster.
+   *
+   * @return The set of known members in the cluster.
+   */
+  Collection<MemberId> getMembers();
 
   /**
    * Returns a list of open sessions.

protocols/raft/src/main/java/io/atomix/protocols/raft/impl/DefaultRaftMetadataClient.java

@@ -61,8 +61,8 @@ public MemberId getLeader() {
   }
 
   @Override
-  public Collection<MemberId> getServers() {
-    return selectorManager.servers();
+  public Collection<MemberId> getMembers() {
+    return selectorManager.members();
   }
 
   /**

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/CommunicationStrategy.java

@@ -38,9 +38,9 @@ public enum  CommunicationStrategy {
    */
   ANY {
     @Override
-    public List<MemberId> selectConnections(MemberId leader, List<MemberId> servers) {
-      Collections.shuffle(servers);
-      return servers;
+    public List<MemberId> selectConnections(MemberId leader, List<MemberId> members) {
+      Collections.shuffle(members);
+      return members;
     }
   },
 
@@ -55,12 +55,12 @@ public List<MemberId> selectConnections(MemberId leader, List<MemberId> servers)
    */
   LEADER {
     @Override
-    public List<MemberId> selectConnections(MemberId leader, List<MemberId> servers) {
+    public List<MemberId> selectConnections(MemberId leader, List<MemberId> members) {
       if (leader != null) {
         return Collections.singletonList(leader);
       }
-      Collections.shuffle(servers);
-      return servers;
+      Collections.shuffle(members);
+      return members;
     }
   },
 
@@ -73,18 +73,18 @@ public List<MemberId> selectConnections(MemberId leader, List<MemberId> servers)
    */
   FOLLOWERS {
     @Override
-    public List<MemberId> selectConnections(MemberId leader, List<MemberId> servers) {
-      Collections.shuffle(servers);
-      if (leader != null && servers.size() > 1) {
-        List<MemberId> results = new ArrayList<>(servers.size());
-        for (MemberId memberId : servers) {
+    public List<MemberId> selectConnections(MemberId leader, List<MemberId> members) {
+      Collections.shuffle(members);
+      if (leader != null && members.size() > 1) {
+        List<MemberId> results = new ArrayList<>(members.size());
+        for (MemberId memberId : members) {
           if (!memberId.equals(leader)) {
             results.add(memberId);
           }
         }
         return results;
       }
-      return servers;
+      return members;
     }
   };
 
@@ -99,11 +99,11 @@ public List<MemberId> selectConnections(MemberId leader, List<MemberId> servers)
    *
    * @param leader  The current cluster leader. The {@code leader} may be {@code null} if no current
    *                leader exists.
-   * @param servers The full list of available servers. The provided server list is representative
+   * @param members The full list of available servers. The provided server list is representative
    *                of the most recent membership update received by the client. The server list
    *                may evolve over time as the structure of the cluster changes.
    * @return A collection of servers to which the client can connect.
    */
-  public abstract List<MemberId> selectConnections(MemberId leader, List<MemberId> servers);
+  public abstract List<MemberId> selectConnections(MemberId leader, List<MemberId> members);
 
 }

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/impl/MemberSelector.java

@@ -57,18 +57,18 @@ public enum State {
 
   private final MemberSelectorManager selectors;
   private MemberId leader;
-  private Collection<MemberId> servers = new LinkedList<>();
+  private Collection<MemberId> members = new LinkedList<>();
   private volatile MemberId selection;
   private final CommunicationStrategy strategy;
   private Collection<MemberId> selections = new LinkedList<>();
   private Iterator<MemberId> selectionsIterator;
 
-  public MemberSelector(MemberId leader, Collection<MemberId> servers, CommunicationStrategy strategy, MemberSelectorManager selectors) {
+  public MemberSelector(MemberId leader, Collection<MemberId> members, CommunicationStrategy strategy, MemberSelectorManager selectors) {
     this.leader = leader;
-    this.servers = checkNotNull(servers, "servers cannot be null");
+    this.members = checkNotNull(members, "servers cannot be null");
     this.strategy = checkNotNull(strategy, "strategy cannot be null");
     this.selectors = checkNotNull(selectors, "selectors cannot be null");
-    this.selections = strategy.selectConnections(leader, new ArrayList<>(servers));
+    this.selections = strategy.selectConnections(leader, new ArrayList<>(members));
   }
 
   /**
@@ -109,34 +109,34 @@ public MemberId leader() {
    *
    * @return The current set of servers.
    */
-  public Collection<MemberId> servers() {
-    return servers;
+  public Collection<MemberId> members() {
+    return members;
   }
 
   /**
-   * Resets the addresses.
+   * Resets the member iterator.
    *
-   * @return The address selector.
+   * @return The member selector.
    */
   public MemberSelector reset() {
     if (selectionsIterator != null) {
-      this.selections = strategy.selectConnections(leader, new ArrayList<>(servers));
+      this.selections = strategy.selectConnections(leader, new ArrayList<>(members));
       this.selectionsIterator = null;
     }
     return this;
   }
 
   /**
-   * Resets the connection addresses.
+   * Resets the connection leader and members.
    *
-   * @param servers The collection of server addresses.
-   * @return The address selector.
+   * @param members The collection of members.
+   * @return The member selector.
    */
-  public MemberSelector reset(MemberId leader, Collection<MemberId> servers) {
-    if (changed(leader, servers)) {
+  public MemberSelector reset(MemberId leader, Collection<MemberId> members) {
+    if (changed(leader, members)) {
       this.leader = leader;
-      this.servers = servers;
-      this.selections = strategy.selectConnections(leader, new ArrayList<>(servers));
+      this.members = members;
+      this.selections = strategy.selectConnections(leader, new ArrayList<>(members));
       this.selectionsIterator = null;
     }
     return this;
@@ -156,7 +156,7 @@ private boolean changed(MemberId leader, Collection<MemberId> servers) {
     } else if (this.leader != null && !this.leader.equals(leader)) {
       checkArgument(servers.contains(leader), "leader must be present in the servers list");
       return true;
-    } else if (!matches(this.servers, servers)) {
+    } else if (!matches(this.members, servers)) {
       return true;
     }
     return false;

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/impl/MemberSelectorManager.java

@@ -30,7 +30,7 @@
 public final class MemberSelectorManager {
   private final Set<MemberSelector> selectors = new CopyOnWriteArraySet<>();
   private volatile MemberId leader;
-  private volatile Collection<MemberId> servers = Collections.emptyList();
+  private volatile Collection<MemberId> members = Collections.emptyList();
 
   /**
    * Returns the current cluster leader.
@@ -42,12 +42,12 @@ public MemberId leader() {
   }
 
   /**
-   * Returns the set of servers in the cluster.
+   * Returns the set of members in the cluster.
    *
-   * @return The set of servers in the cluster.
+   * @return The set of members in the cluster.
    */
-  public Collection<MemberId> servers() {
-    return servers;
+  public Collection<MemberId> members() {
+    return members;
   }
 
   /**
@@ -57,7 +57,7 @@ public Collection<MemberId> servers() {
    * @return A new address selector.
    */
   public MemberSelector createSelector(CommunicationStrategy selectionStrategy) {
-    MemberSelector selector = new MemberSelector(leader, servers, selectionStrategy, this);
+    MemberSelector selector = new MemberSelector(leader, members, selectionStrategy, this);
     selectors.add(selector);
     return selector;
   }
@@ -73,18 +73,18 @@ public void resetAll() {
    * Resets all child selectors.
    *
    * @param leader  The current cluster leader.
-   * @param servers The collection of all active servers.
+   * @param members The collection of all active members.
    */
-  public void resetAll(MemberId leader, Collection<MemberId> servers) {
+  public void resetAll(MemberId leader, Collection<MemberId> members) {
     this.leader = leader;
-    this.servers = new LinkedList<>(servers);
-    selectors.forEach(s -> s.reset(leader, servers));
+    this.members = new LinkedList<>(members);
+    selectors.forEach(s -> s.reset(leader, members));
   }
 
   /**
    * Removes the given selector.
    *
-   * @param selector The address selector to remove.
+   * @param selector The member selector to remove.
    */
   void remove(MemberSelector selector) {
     selectors.remove(selector);

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/impl/RaftProxyConnection.java

@@ -75,6 +75,23 @@ public RaftProxyConnection(RaftClientProtocol protocol, MemberSelector selector,
     this.log = ContextualLoggerFactory.getLogger(getClass(), loggerContext);
   }
 
+  /**
+   * Resets the member selector.
+   */
+  public void reset() {
+    selector.reset();
+  }
+
+  /**
+   * Resets the member selector.
+   *
+   * @param leader the selector leader
+   * @param servers the selector servers
+   */
+  public void reset(MemberId leader, Collection<MemberId> servers) {
+    selector.reset(leader, servers);
+  }
+
   /**
    * Returns the current selector leader.
    *
@@ -85,12 +102,12 @@ public MemberId leader() {
   }
 
   /**
-   * Returns the current set of servers.
+   * Returns the current set of members.
    *
-   * @return The current set of servers.
+   * @return The current set of members.
    */
-  public Collection<MemberId> servers() {
-    return selector.servers();
+  public Collection<MemberId> members() {
+    return selector.members();
   }
 
   /**

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/impl/RaftProxyInvoker.java

@@ -361,6 +361,9 @@ else if (response.error().type() == RaftError.Type.UNKNOWN_CLIENT
           retry(Duration.ofSeconds(FIBONACCI[Math.min(attempt - 1, FIBONACCI.length - 1)]));
         }
       } else if (EXCEPTION_PREDICATE.test(error) || (error instanceof CompletionException && EXCEPTION_PREDICATE.test(error.getCause()))) {
+        if (error instanceof ConnectException || error.getCause() instanceof ConnectException) {
+          leaderConnection.reset(null, leaderConnection.members());
+        }
         retry(Duration.ofSeconds(FIBONACCI[Math.min(attempt - 1, FIBONACCI.length - 1)]));
       } else {
         fail(error);

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/impl/RaftProxyListener.java

@@ -111,7 +111,7 @@ private void handlePublish(PublishRequest request) {
           .withIndex(eventIndex)
           .build();
       log.trace("Sending {}", resetRequest);
-      protocol.reset(memberSelector.servers(), resetRequest);
+      protocol.reset(memberSelector.members(), resetRequest);
       return;
     }
 

protocols/raft/src/main/java/io/atomix/protocols/raft/proxy/impl/RaftProxyManager.java

@@ -317,7 +317,7 @@ private synchronized void keepAliveSessions(long lastKeepAliveTime, long session
           // If the timeout has not been passed, attempt to keep the session alive again with no delay.
           // We will continue to retry until the session expiration has passed.
           else if (System.currentTimeMillis() - lastKeepAliveTime < sessionTimeout) {
-            selectorManager.resetAll(null, connection.servers());
+            selectorManager.resetAll(null, connection.members());
             keepAliveSessions(lastKeepAliveTime, sessionTimeout);
           }
           // If no leader was set, set the session state to unstable and schedule another keep-alive.
@@ -330,7 +330,7 @@ else if (System.currentTimeMillis() - lastKeepAliveTime < sessionTimeout) {
         // If the timeout has not been passed, reset the connection and attempt to keep the session alive
         // again with no delay.
         else if (System.currentTimeMillis() - lastKeepAliveTime < sessionTimeout && connection.leader() != null) {
-          selectorManager.resetAll(null, connection.servers());
+          selectorManager.resetAll(null, connection.members());
           keepAliveSessions(lastKeepAliveTime, sessionTimeout);
         }
         // If no leader was set, set the session state to unstable and schedule another keep-alive.

protocols/raft/src/test/java/io/atomix/protocols/raft/proxy/impl/MemberSelectorManagerTest.java

@@ -35,13 +35,13 @@ public class MemberSelectorManagerTest {
   public void testMemberSelectorManager() throws Exception {
     MemberSelectorManager selectorManager = new MemberSelectorManager();
     assertNull(selectorManager.leader());
-    assertEquals(0, selectorManager.servers().size());
+    assertEquals(0, selectorManager.members().size());
     selectorManager.resetAll();
     assertNull(selectorManager.leader());
-    assertEquals(0, selectorManager.servers().size());
+    assertEquals(0, selectorManager.members().size());
     selectorManager.resetAll(MemberId.from("a"), Arrays.asList(MemberId.from("a"), MemberId.from("b"), MemberId.from("c")));
     assertEquals(MemberId.from("a"), selectorManager.leader());
-    assertEquals(3, selectorManager.servers().size());
+    assertEquals(3, selectorManager.members().size());
   }
 
 }

protocols/raft/src/test/java/io/atomix/protocols/raft/proxy/impl/MemberSelectorTest.java

@@ -133,7 +133,7 @@ public void testSelectLeader() throws Exception {
 
     selectorManager.resetAll(MemberId.from("a"), Arrays.asList(MemberId.from("a"), MemberId.from("b"), MemberId.from("c")));
     assertEquals(MemberId.from("a"), selector.leader());
-    assertEquals(3, selector.servers().size());
+    assertEquals(3, selector.members().size());
     assertTrue(selector.hasNext());
     assertNotNull(selector.next());
     assertFalse(selector.hasNext());