Graceful Shutdown of Edge Servers

When edge servers shutdown gracefully, they remove themselves from the
cluster topology.

enterprise/core-edge/src/main/java/org/neo4j/coreedge/catchup/CoreClient.java

@@ -19,14 +19,12 @@
  */
 package org.neo4j.coreedge.catchup;
 
-import io.netty.channel.ChannelInitializer;
-import io.netty.channel.socket.SocketChannel;
-
 import java.util.concurrent.CompletableFuture;
 import java.util.function.Consumer;
 
-import org.neo4j.coreedge.core.state.snapshot.CoreSnapshotListener;
-import org.neo4j.coreedge.core.state.snapshot.CoreSnapshotRequest;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.socket.SocketChannel;
+
 import org.neo4j.coreedge.catchup.storecopy.GetStoreIdRequest;
 import org.neo4j.coreedge.catchup.storecopy.GetStoreRequest;
 import org.neo4j.coreedge.catchup.storecopy.StoreFileReceiver;
@@ -38,15 +36,17 @@
 import org.neo4j.coreedge.catchup.tx.TxPullResponse;
 import org.neo4j.coreedge.catchup.tx.TxPullResponseListener;
 import org.neo4j.coreedge.catchup.tx.TxStreamCompleteListener;
-import org.neo4j.coreedge.discovery.TopologyService;
-import org.neo4j.coreedge.messaging.Message;
-import org.neo4j.coreedge.messaging.CoreOutbound;
-import org.neo4j.coreedge.messaging.Outbound;
 import org.neo4j.coreedge.core.state.snapshot.CoreSnapshot;
+import org.neo4j.coreedge.core.state.snapshot.CoreSnapshotListener;
+import org.neo4j.coreedge.core.state.snapshot.CoreSnapshotRequest;
+import org.neo4j.coreedge.discovery.TopologyService;
 import org.neo4j.coreedge.identity.MemberId;
+import org.neo4j.coreedge.identity.StoreId;
+import org.neo4j.coreedge.messaging.CoreOutbound;
+import org.neo4j.coreedge.messaging.Message;
 import org.neo4j.coreedge.messaging.NonBlockingChannels;
+import org.neo4j.coreedge.messaging.Outbound;
 import org.neo4j.coreedge.messaging.SenderService;
-import org.neo4j.coreedge.identity.StoreId;
 import org.neo4j.helpers.Listeners;
 import org.neo4j.kernel.lifecycle.LifecycleAdapter;
 import org.neo4j.kernel.monitoring.Monitors;
@@ -55,9 +55,9 @@
 import static java.util.Arrays.asList;
 
 public abstract class CoreClient extends LifecycleAdapter implements StoreFileReceiver, StoreIdReceiver,
-                                                                     StoreFileStreamingCompleteListener,
-                                                                     TxStreamCompleteListener, TxPullResponseListener,
-                                                                     CoreSnapshotListener
+        StoreFileStreamingCompleteListener,
+        TxStreamCompleteListener, TxPullResponseListener,
+        CoreSnapshotListener
 {
     private final PullRequestMonitor pullRequestMonitor;
     private final SenderService senderService;
@@ -71,8 +71,8 @@ public abstract class CoreClient extends LifecycleAdapter implements StoreFileRe
     private Outbound<MemberId, Message> outbound;
 
     public CoreClient( LogProvider logProvider, ChannelInitializer<SocketChannel> channelInitializer, Monitors monitors,
-            int maxQueueSize, NonBlockingChannels nonBlockingChannels, TopologyService discoveryService,
-            long logThresholdMillis )
+                       int maxQueueSize, NonBlockingChannels nonBlockingChannels, TopologyService discoveryService,
+                       long logThresholdMillis )
     {
         senderService =
                 new SenderService( channelInitializer, logProvider, monitors, maxQueueSize, nonBlockingChannels );
@@ -109,7 +109,7 @@ public void pollForTransactions( MemberId serverAddress, StoreId storeId, long l
 
     private void send( MemberId to, RequestMessageType messageType, Message contentMessage )
     {
-        outbound.send( to,  asList( messageType, contentMessage ) );
+        outbound.send( to, asList( messageType, contentMessage ) );
     }
 
     @Override

enterprise/core-edge/src/main/java/org/neo4j/coreedge/core/CoreEdgeClusterSettings.java

@@ -142,8 +142,8 @@ public String toString()
             setting( "core_edge.discovery_listen_address", LISTEN_SOCKET_ADDRESS, "0.0.0.0:5000" );
 
     @Description("A comma-separated list of other members of the cluster to join.")
-    public static final Setting<List<AdvertisedSocketAddress>> initial_core_cluster_members =
-            setting( "core_edge.initial_core_cluster_members", list( ",", ADVERTISED_SOCKET_ADDRESS ), MANDATORY );
+    public static final Setting<List<AdvertisedSocketAddress>> initial_hazelcast_members =
+            setting( "core_edge.initial_hazelcast_members", list( ",", ADVERTISED_SOCKET_ADDRESS ), MANDATORY );
 
     @Description("Prevents the network middleware from dumping its own logs. Defaults to true.")
     public static final Setting<Boolean> disable_middleware_logging =

enterprise/core-edge/src/main/java/org/neo4j/coreedge/core/CoreGraphDatabase.java

@@ -53,7 +53,7 @@ public CoreGraphDatabase( File storeDir, Map<String, String> params,
 
     public MemberId id()
     {
-        return (MemberId) getDependencyResolver().resolveDependency( RaftMachine.class ).identity();
+        return getDependencyResolver().resolveDependency( RaftMachine.class ).identity();
     }
 
     public Role getRole()

enterprise/core-edge/src/main/java/org/neo4j/coreedge/discovery/ClusterTopology.java

@@ -51,7 +51,7 @@ public Collection<CoreAddresses> coreMemberAddresses()
         return coreMembers.values();
     }
 
-    public Set<EdgeAddresses> edgeMemberAddresses()
+    public Collection<EdgeAddresses> edgeMemberAddresses()
     {
         return edgeAddresses;
     }

enterprise/core-edge/src/main/java/org/neo4j/coreedge/discovery/HazelcastClient.java

@@ -54,7 +54,7 @@ public ClusterTopology currentTopology()
         try
         {
             return retry( ( hazelcastInstance ) ->
-                    HazelcastClusterTopology.fromHazelcastInstance( hazelcastInstance, log ) );
+                    HazelcastClusterTopology.getClusterTopology( hazelcastInstance, log ) );
         }
         catch ( Exception e )
         {
@@ -76,6 +76,9 @@ public synchronized void stop() throws Throwable
     {
         if ( hazelcastInstance != null )
         {
+            hazelcastInstance.getMap( EDGE_SERVER_BOLT_ADDRESS_MAP_NAME )
+                    .remove( hazelcastInstance.getLocalEndpoint().getUuid() );
+
             hazelcastInstance.shutdown();
         }
     }

enterprise/core-edge/src/main/java/org/neo4j/coreedge/discovery/HazelcastClientConnector.java

@@ -21,6 +21,7 @@
 
 import com.hazelcast.client.HazelcastClient;
 import com.hazelcast.client.config.ClientConfig;
+import com.hazelcast.client.config.ClientNetworkConfig;
 import com.hazelcast.core.HazelcastInstance;
 
 import org.neo4j.coreedge.messaging.address.AdvertisedSocketAddress;
@@ -43,7 +44,7 @@ public HazelcastInstance connectToHazelcast()
 
         clientConfig.getGroupConfig().setName( config.get( CoreEdgeClusterSettings.cluster_name ) );
 
-        for ( AdvertisedSocketAddress address : config.get( CoreEdgeClusterSettings.initial_core_cluster_members ) )
+        for ( AdvertisedSocketAddress address : config.get( CoreEdgeClusterSettings.initial_hazelcast_members ) )
         {
             clientConfig.getNetworkConfig().addAddress( address.toString() );
         }

enterprise/core-edge/src/main/java/org/neo4j/coreedge/discovery/HazelcastClusterTopology.java

@@ -19,14 +19,6 @@
  */
 package org.neo4j.coreedge.discovery;
 
-import com.hazelcast.config.MemberAttributeConfig;
-import com.hazelcast.core.Client;
-import com.hazelcast.core.ClientService;
-import com.hazelcast.core.HazelcastInstance;
-import com.hazelcast.core.HazelcastInstanceAware;
-import com.hazelcast.core.IExecutorService;
-import com.hazelcast.core.Member;
-
 import java.io.Serializable;
 import java.util.Collection;
 import java.util.HashMap;
@@ -40,6 +32,14 @@
 import java.util.concurrent.Future;
 import java.util.stream.Collectors;
 
+import com.hazelcast.config.MemberAttributeConfig;
+import com.hazelcast.core.Client;
+import com.hazelcast.core.ClientService;
+import com.hazelcast.core.HazelcastInstance;
+import com.hazelcast.core.HazelcastInstanceAware;
+import com.hazelcast.core.IExecutorService;
+import com.hazelcast.core.Member;
+
 import org.neo4j.coreedge.core.CoreEdgeClusterSettings;
 import org.neo4j.coreedge.edge.EnterpriseEdgeEditionModule;
 import org.neo4j.coreedge.identity.MemberId;
@@ -53,13 +53,14 @@
 
 class HazelcastClusterTopology
 {
-    static final String EDGE_SERVER_BOLT_ADDRESS_MAP_NAME = "edge-servers"; // hz client uuid string -> boltAddress string
+    // hz client uuid string -> boltAddress string
+    static final String EDGE_SERVER_BOLT_ADDRESS_MAP_NAME = "edge-servers";
     static final String MEMBER_UUID = "member_uuid";
     static final String TRANSACTION_SERVER = "transaction_server";
     static final String RAFT_SERVER = "raft_server";
     static final String BOLT_SERVER = "bolt_server";
 
-    static ClusterTopology fromHazelcastInstance( HazelcastInstance hazelcastInstance, Log log )
+    static ClusterTopology getClusterTopology( HazelcastInstance hazelcastInstance, Log log )
     {
         Set<Member> coreMembers = emptySet();
         if ( hazelcastInstance != null )
@@ -70,22 +71,29 @@ static ClusterTopology fromHazelcastInstance( HazelcastInstance hazelcastInstanc
                 edgeMembers( hazelcastInstance ) );
     }
 
-    private static class GetConnectedClients implements Callable<Collection<String>>, HazelcastInstanceAware, Serializable
+    static class GetConnectedClients implements Callable<Collection<String>>, Serializable, HazelcastInstanceAware
     {
         private transient HazelcastInstance instance;
 
+        GetConnectedClients( HazelcastInstance instance )
+        {
+            this.instance = instance;
+        }
+
         @Override
         public Collection<String> call() throws Exception
         {
             final ClientService clientService = instance.getClientService();
-            final Collection<Client> connectedClients = clientService.getConnectedClients();
-            return connectedClients.stream().map( Client::getUuid ).collect( Collectors.toCollection( HashSet::new ) );
+            return clientService.getConnectedClients()
+                    .stream()
+                    .map( Client::getUuid )
+                    .collect( Collectors.toCollection( HashSet::new ) );
         }
 
         @Override
         public void setHazelcastInstance( HazelcastInstance hazelcastInstance )
         {
-            instance = hazelcastInstance;
+            this.instance = hazelcastInstance;
         }
     }
 
@@ -101,19 +109,17 @@ private static Set<EdgeAddresses> edgeMembers( HazelcastInstance hazelcastInstan
         final IExecutorService executorService = hazelcastInstance.getExecutorService( "default" );
         try
         {
-            GetConnectedClients getConnectedClients = new GetConnectedClients();
-            getConnectedClients.setHazelcastInstance( hazelcastInstance );
-            Future<Collection<String>> collectionFuture = executorService.submit( getConnectedClients );
-            connectedUUIDs = collectionFuture.get();
+            connectedUUIDs = executorService.submit( new GetConnectedClients( hazelcastInstance ) ).get();
         }
         catch ( InterruptedException | ExecutionException e )
         {
             // todo log a warning
             return emptySet();
         }
 
-        return hazelcastInstance.<String/*uuid*/,String/*boltAddress*/>getMap( EDGE_SERVER_BOLT_ADDRESS_MAP_NAME ).entrySet().stream().
-                filter( entry -> connectedUUIDs.contains( entry.getKey() ) )
+        return hazelcastInstance.<String/*uuid*/, String/*boltAddress*/>getMap( EDGE_SERVER_BOLT_ADDRESS_MAP_NAME )
+                .entrySet().stream().
+                        filter( entry -> connectedUUIDs.contains( entry.getKey() ) )
                 .map( entry -> new EdgeAddresses( new AdvertisedSocketAddress( entry.getValue() /*boltAddress*/ ) ) )
                 .collect( toSet() );
     }
@@ -124,15 +130,15 @@ private static boolean canBeBootstrapped( Set<Member> coreMembers )
         return iterator.hasNext() && iterator.next().localMember();
     }
 
-    static Map<MemberId,CoreAddresses> toCoreMemberMap( Set<Member> members, Log log )
+    static Map<MemberId, CoreAddresses> toCoreMemberMap( Set<Member> members, Log log )
     {
-        Map<MemberId,CoreAddresses> coreMembers = new HashMap<>();
+        Map<MemberId, CoreAddresses> coreMembers = new HashMap<>();
 
         for ( Member member : members )
         {
             try
             {
-                Pair<MemberId,CoreAddresses> pair = extractMemberAttributes( member );
+                Pair<MemberId, CoreAddresses> pair = extractMemberAttributes( member );
                 coreMembers.put( pair.first(), pair.other() );
             }
             catch ( IllegalArgumentException e )
@@ -149,7 +155,8 @@ static MemberAttributeConfig buildMemberAttributes( MemberId myself, Config conf
         MemberAttributeConfig memberAttributeConfig = new MemberAttributeConfig();
         memberAttributeConfig.setStringAttribute( MEMBER_UUID, myself.getUuid().toString() );
 
-        AdvertisedSocketAddress transactionSource = config.get( CoreEdgeClusterSettings.transaction_advertised_address );
+        AdvertisedSocketAddress transactionSource = config.get( CoreEdgeClusterSettings
+                .transaction_advertised_address );
         memberAttributeConfig.setStringAttribute( TRANSACTION_SERVER, transactionSource.toString() );
 
         AdvertisedSocketAddress raftAddress = config.get( CoreEdgeClusterSettings.raft_advertised_address );
@@ -160,7 +167,7 @@ static MemberAttributeConfig buildMemberAttributes( MemberId myself, Config conf
         return memberAttributeConfig;
     }
 
-    static Pair<MemberId,CoreAddresses> extractMemberAttributes( Member member )
+    static Pair<MemberId, CoreAddresses> extractMemberAttributes( Member member )
     {
         MemberId memberId = new MemberId( UUID.fromString( member.getStringAttribute( MEMBER_UUID ) ) );
 

enterprise/core-edge/src/main/java/org/neo4j/coreedge/discovery/HazelcastCoreTopologyService.java

@@ -120,7 +120,7 @@ private HazelcastInstance createHazelcastInstance()
         tcpIpConfig.setEnabled( true );
 
         List<AdvertisedSocketAddress> initialMembers =
-                config.get( CoreEdgeClusterSettings.initial_core_cluster_members );
+                config.get( CoreEdgeClusterSettings.initial_hazelcast_members );
         for ( AdvertisedSocketAddress address : initialMembers )
         {
             tcpIpConfig.addMember( address.toString() );
@@ -155,6 +155,6 @@ private Integer minimumClusterSizeThatCanTolerateOneFaultForExpectedClusterSize(
     @Override
     public ClusterTopology currentTopology()
     {
-        return HazelcastClusterTopology.fromHazelcastInstance( hazelcastInstance, log );
+        return HazelcastClusterTopology.getClusterTopology( hazelcastInstance, log );
     }
 }