Ensure tryLock attempts are rejected during network partitions.

- Use client-side timers for tryLock with timeout
- Ensure lock sessions are properly cleaned up when timed out by clients

core/src/main/java/io/atomix/core/Atomix.java

@@ -78,6 +78,7 @@
 import io.atomix.utils.concurrent.Futures;
 import io.atomix.utils.concurrent.SingleThreadContext;
 import io.atomix.utils.concurrent.ThreadContext;
+import io.atomix.utils.concurrent.Threads;
 import io.atomix.utils.net.Address;
 import io.atomix.utils.net.MalformedAddressException;
 import org.slf4j.Logger;
@@ -89,6 +90,8 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.stream.Collectors;
 
@@ -423,6 +426,7 @@ private synchronized CompletableFuture<Void> doStop() {
         .thenComposeAsync(v -> context.messagingService.stop(), threadContext)
         .exceptionally(e -> null)
         .thenRunAsync(() -> {
+          context.executorService.shutdownNow();
           threadContext.close();
           started.set(false);
           LOGGER.info("Stopped");
@@ -474,6 +478,7 @@ private static AtomixConfig loadConfig(File config) {
    * Builds a context from the given configuration.
    */
   private static Context buildContext(AtomixConfig config) {
+    ScheduledExecutorService executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), Threads.namedThreads("atomix-primitive-%d", LOGGER));
     ManagedMessagingService messagingService = buildMessagingService(config);
     ManagedBroadcastService broadcastService = buildBroadcastService(config);
     ManagedBootstrapMetadataService bootstrapMetadataService = buildBootstrapMetadataService(config);
@@ -484,9 +489,10 @@ private static Context buildContext(AtomixConfig config) {
     ManagedPartitionGroup systemPartitionGroup = buildSystemPartitionGroup(config);
     ManagedPartitionService partitions = buildPartitionService(config);
     ManagedPrimitivesService primitives = new CorePrimitivesService(
-        clusterService, clusterMessagingService, clusterEventingService, partitions, systemPartitionGroup, config);
+        executorService, clusterService, clusterMessagingService, clusterEventingService, partitions, systemPartitionGroup, config);
     PrimitiveTypeRegistry primitiveTypes = new PrimitiveTypeRegistry(config.getPrimitiveTypes());
     return new Context(
+        executorService,
         messagingService,
         broadcastService,
         bootstrapMetadataService,
@@ -618,6 +624,7 @@ private static ManagedPartitionService buildPartitionService(AtomixConfig config
    * Atomix instance context.
    */
   private static class Context {
+    private final ScheduledExecutorService executorService;
     private final ManagedMessagingService messagingService;
     private final ManagedBroadcastService broadcastService;
     private final ManagedBootstrapMetadataService bootstrapMetadataService;
@@ -632,6 +639,7 @@ private static class Context {
     private final boolean enableShutdownHook;
 
     public Context(
+        ScheduledExecutorService executorService,
         ManagedMessagingService messagingService,
         ManagedBroadcastService broadcastService,
         ManagedBootstrapMetadataService bootstrapMetadataService,
@@ -644,6 +652,7 @@ public Context(
         ManagedPrimitivesService primitives,
         PrimitiveTypeRegistry primitiveTypes,
         boolean enableShutdownHook) {
+      this.executorService = executorService;
       this.messagingService = messagingService;
       this.broadcastService = broadcastService;
       this.bootstrapMetadataService = bootstrapMetadataService;
@@ -872,6 +881,7 @@ public Builder addPrimitiveType(PrimitiveType primitiveType) {
      */
     @Override
     public Atomix build() {
+      ScheduledExecutorService executorService = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), Threads.namedThreads("atomix-primitive-%d", LOGGER));
       ManagedMessagingService messagingService = buildMessagingService();
       ManagedBroadcastService broadcastService = buildBroadcastService();
       ManagedBootstrapMetadataService bootstrapMetadataService = buildBootstrapMetadataService();
@@ -882,8 +892,9 @@ public Atomix build() {
       ManagedPartitionGroup systemPartitionGroup = buildSystemPartitionGroup();
       ManagedPartitionService partitions = buildPartitionService();
       ManagedPrimitivesService primitives = new CorePrimitivesService(
-          clusterService, clusterMessagingService, clusterEventingService, partitions, systemPartitionGroup, new AtomixConfig());
+          executorService, clusterService, clusterMessagingService, clusterEventingService, partitions, systemPartitionGroup, new AtomixConfig());
       return new Atomix(new Context(
+          executorService,
           messagingService,
           broadcastService,
           bootstrapMetadataService,

core/src/main/java/io/atomix/core/impl/CorePrimitiveManagementService.java

@@ -22,29 +22,39 @@
 import io.atomix.primitive.PrimitiveRegistry;
 import io.atomix.primitive.partition.PartitionService;
 
+import java.util.concurrent.ScheduledExecutorService;
+
 /**
  * Default primitive management service.
  */
 public class CorePrimitiveManagementService implements PrimitiveManagementService {
+  private final ScheduledExecutorService executorService;
   private final ClusterService clusterService;
   private final ClusterMessagingService communicationService;
   private final ClusterEventingService eventService;
   private final PartitionService partitionService;
   private final PrimitiveRegistry primitiveRegistry;
 
   public CorePrimitiveManagementService(
+      ScheduledExecutorService executorService,
       ClusterService clusterService,
       ClusterMessagingService communicationService,
       ClusterEventingService eventService,
       PartitionService partitionService,
       PrimitiveRegistry primitiveRegistry) {
+    this.executorService = executorService;
     this.clusterService = clusterService;
     this.communicationService = communicationService;
     this.eventService = eventService;
     this.partitionService = partitionService;
     this.primitiveRegistry = primitiveRegistry;
   }
 
+  @Override
+  public ScheduledExecutorService getExecutorService() {
+    return executorService;
+  }
+
   @Override
   public ClusterService getClusterService() {
     return clusterService;

core/src/main/java/io/atomix/core/impl/CorePrimitivesService.java

@@ -67,6 +67,7 @@
 import java.util.Collection;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static com.google.common.base.Preconditions.checkNotNull;
@@ -87,6 +88,7 @@ public class CorePrimitivesService implements ManagedPrimitivesService {
   private final AtomicBoolean started = new AtomicBoolean();
 
   public CorePrimitivesService(
+      ScheduledExecutorService executorService,
       ClusterService clusterService,
       ClusterMessagingService communicationService,
       ClusterEventingService eventService,
@@ -95,6 +97,7 @@ public CorePrimitivesService(
       AtomixConfig config) {
     this.primitiveRegistry = new CorePrimitiveRegistry(systemPartitionGroup);
     this.managementService = new CorePrimitiveManagementService(
+        executorService,
         clusterService,
         communicationService,
         eventService,

core/src/main/java/io/atomix/core/lock/impl/DistributedLockEvents.java

@@ -23,8 +23,8 @@
  * Raft value events.
  */
 public enum DistributedLockEvents implements EventType {
-  LOCK("lock"),
-  FAIL("fail");
+  LOCKED("lock"),
+  FAILED("fail");
 
   private final String id;