Reap consul sessions when releasing lock; add session TTL

soabase · Apr 12, 2017 · a5e9cfa · a5e9cfa
1 parent d746911
commit a5e9cfa
Show file tree

Hide file tree

Showing 2 changed files with 47 additions and 10 deletions.
diff --git a/exhibitor-core/src/main/java/com/netflix/exhibitor/core/config/consul/ConsulKvLock.java b/exhibitor-core/src/main/java/com/netflix/exhibitor/core/config/consul/ConsulKvLock.java
@@ -14,26 +14,48 @@ public class ConsulKvLock {
     private final Consul consul;
     private final String path;
     private final String name;
+    private final String ttl;
+    private String sessionId;
 
     /**
      * @param consul consul client instance for connecting to consul cluster
      * @param path consul key-value path to lock
      * @param name a descriptive name for the lock
+     * @param ttl TTL, in seconds, for the consul session underpinning the lock
      */
-    public ConsulKvLock(Consul consul, String path, String name) {
+    public ConsulKvLock(Consul consul, String path, String name, Integer ttl) {
         this.consul = consul;
         this.path = path;
         this.name = name;
+        this.ttl = ttl != null ? String.format("%ds", ttl) : null;
+        this.sessionId = null;
+    }
+
+    /**
+     * @param consul consul client instance for connecting to consul cluster
+     * @param path consul key-value path to lock
+     * @param name a descriptive name for the lock
+     */
+    public ConsulKvLock(Consul consul, String path, String name) {
+        this(consul, path, name, 60);
     }
 
     private String createSession() {
-        final ImmutableSession session = ImmutableSession.builder().name(name).build();
+        final ImmutableSession session = ImmutableSession.builder()
+                .name(name)
+                .ttl(Optional.fromNullable(ttl))
+                .build();
         return consul.sessionClient().createSession(session).getId();
     }
 
+    private void destroySession() {
+        consul.sessionClient().destroySession(sessionId);
+        sessionId = null;
+    }
+
     public boolean acquireLock(long maxWait, TimeUnit unit) {
         KeyValueClient kv = consul.keyValueClient();
-        String sessionId = createSession();
+        sessionId = createSession();
 
         Optional<Value> value = kv.getValue(path);
 
@@ -43,18 +65,28 @@ public boolean acquireLock(long maxWait, TimeUnit unit) {
 
         BigInteger index = BigInteger.valueOf(value.get().getModifyIndex());
         kv.getValue(path, QueryOptions.blockMinutes((int) unit.toMinutes(maxWait), index).build());
-        return kv.acquireLock(path, sessionId);
+
+        if (!kv.acquireLock(path, sessionId)) {
+            destroySession();
+            return false;
+        } else {
+            return true;
+        }
     }
 
     public void releaseLock() {
-        KeyValueClient kv = consul.keyValueClient();
-        Optional<Value> value = kv.getValue(path);
+        try {
+            KeyValueClient kv = consul.keyValueClient();
+            Optional<Value> value = kv.getValue(path);
 
-        if (value.isPresent()) {
-            Optional<String> session = value.get().getSession();
-            if (session.isPresent()) {
-                kv.releaseLock(path, session.get());
+            if (value.isPresent()) {
+                Optional<String> session = value.get().getSession();
+                if (session.isPresent()) {
+                    kv.releaseLock(path, session.get());
+                }
             }
+        } finally {
+            destroySession();
         }
     }
 }
diff --git a/...core/src/test/java/com/netflix/exhibitor/core/config/consul/TestConsulConfigProvider.java b/...core/src/test/java/com/netflix/exhibitor/core/config/consul/TestConsulConfigProvider.java
@@ -5,13 +5,15 @@
 import com.netflix.exhibitor.core.config.PropertyBasedInstanceConfig;
 import com.netflix.exhibitor.core.config.StringConfigs;
 import com.orbitz.consul.Consul;
+import com.orbitz.consul.model.session.SessionInfo;
 import com.pszymczyk.consul.ConsulProcess;
 import com.pszymczyk.consul.ConsulStarterBuilder;
 import org.apache.curator.test.Timing;
 import org.apache.curator.utils.CloseableUtils;
 import org.testng.Assert;
 import org.testng.annotations.*;
 
+import java.util.List;
 import java.util.Properties;
 
 
@@ -57,6 +59,9 @@ public void testBasic() throws Exception {
 
             LoadedInstanceConfig instanceConfig = config.loadConfig();
             Assert.assertEquals(instanceConfig.getConfig().getRootConfig().getString(StringConfigs.ZOO_CFG_EXTRA), "1,2,3");
+
+            List<SessionInfo> sessions = client.sessionClient().listSessions();
+            Assert.assertEquals(sessions.size(), 0, "Consul session still exists!");
         }
         finally {
             CloseableUtils.closeQuietly(config);