Add support for multiple cluster-ids in one setup

Author: jdhoek

uniqueid-etcd/src/main/java/org/lable/oss/uniqueid/etcd/ClusterID.java

@@ -22,7 +22,11 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.List;
 import java.util.concurrent.ExecutionException;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 public class ClusterID {
     final static ByteSequence CLUSTER_ID_KEY = ByteSequence.from("cluster-id", StandardCharsets.UTF_8);
@@ -35,34 +39,46 @@ public class ClusterID {
      * @return The cluster ID, if configured in the cluster.
      * @throws IOException Thrown when retrieving the ID fails.
      */
-    public static int get(Client etcd) throws IOException {
+    public static List<Integer> get(Client etcd) throws IOException {
         GetResponse get;
         try {
             get = etcd.getKVClient().get(CLUSTER_ID_KEY).get();
         } catch (InterruptedException | ExecutionException e) {
             throw new IOException(e);
         }
 
-        Integer id = null;
+        List<Integer> ids = null;
 
         for (KeyValue kv : get.getKvs()) {
             if (kv.getKey().equals(CLUSTER_ID_KEY)) {
+                // There should be only one key returned.
                 String value = kv.getValue().toString(StandardCharsets.UTF_8);
-                id = Integer.parseInt(value);
+                try {
+                    ids = parseIntegers(value);
+                } catch (NumberFormatException e) {
+                    throw new IOException("Failed to parse cluster-id value `" + value + "`.", e);
+                }
                 break;
             }
         }
 
-        if (id == null) {
+        if (ids == null) {
             ByteSequence defaultValue = ByteSequence.from(String.valueOf(DEFAULT_CLUSTER_ID).getBytes());
             try {
                 etcd.getKVClient().put(CLUSTER_ID_KEY, defaultValue).get();
-                return DEFAULT_CLUSTER_ID;
+                return Collections.singletonList(DEFAULT_CLUSTER_ID);
             } catch (InterruptedException | ExecutionException e) {
                 throw new IOException(e);
             }
         } else {
-            return id;
+            return ids;
         }
     }
+
+    static List<Integer> parseIntegers(String serialized) {
+        return Stream.of(serialized.split(","))
+                .map(String::trim)
+                .map(Integer::parseInt)
+                .collect(Collectors.toList());
+    }
 }

uniqueid-etcd/src/main/java/org/lable/oss/uniqueid/etcd/ExpiringResourceClaim.java

@@ -21,6 +21,7 @@
 
 import java.io.IOException;
 import java.time.Duration;
+import java.util.List;
 import java.util.Timer;
 import java.util.TimerTask;
 
@@ -35,10 +36,11 @@ public class ExpiringResourceClaim extends ResourceClaim {
     public final static Duration DEFAULT_ACQUISITION_TIMEOUT = Duration.ofMinutes(10);
 
     ExpiringResourceClaim(Client etcd,
-                          int poolSize,
+                          int maxGeneratorCount,
+                          List<Integer> clusterIds,
                           Duration claimHold,
                           Duration acquisitionTimeout) throws IOException {
-        super(etcd, poolSize, acquisitionTimeout);
+        super(etcd, maxGeneratorCount, clusterIds, acquisitionTimeout);
         new Timer().schedule(new TimerTask() {
             @Override
             public void run() {
@@ -50,28 +52,30 @@ public void run() {
     /**
      * Claim a resource.
      *
-     * @param etcd     Etcd connection to use.
-     * @param poolSize Size of the resource pool.
+     * @param etcd              Etcd connection to use.
+     * @param maxGeneratorCount Maximum number of generators possible.
      * @return A resource claim.
      */
-    public static ResourceClaim claimExpiring(Client etcd, int poolSize)
+    public static ResourceClaim claimExpiring(Client etcd, int maxGeneratorCount, List<Integer> clusterIds)
             throws IOException {
-        return claimExpiring(etcd, poolSize, DEFAULT_CLAIM_HOLD, DEFAULT_ACQUISITION_TIMEOUT);
+        return claimExpiring(etcd, maxGeneratorCount, clusterIds, DEFAULT_CLAIM_HOLD, DEFAULT_ACQUISITION_TIMEOUT);
     }
 
     /**
      * Claim a resource.
      *
      * @param etcd               Etcd connection to use.
-     * @param poolSize           Size of the resource pool.
+     * @param maxGeneratorCount  Maximum number of generators possible.
+     * @param clusterIds         Cluster Ids available to use.
      * @param claimHold          How long the claim should be held. May be {@code null} for the default value of
      *                           {@link #DEFAULT_CLAIM_HOLD}.
      * @param acquisitionTimeout How long to keep trying to acquire a claim. May be {@code null} to keep trying
      *                           indefinitely.
      * @return A resource claim.
      */
     public static ResourceClaim claimExpiring(Client etcd,
-                                              int poolSize,
+                                              int maxGeneratorCount,
+                                              List<Integer> clusterIds,
                                               Duration claimHold,
                                               Duration acquisitionTimeout)
             throws IOException {
@@ -81,6 +85,6 @@ public static ResourceClaim claimExpiring(Client etcd,
             logger.debug("Preparing expiring resource-claim; will release it in {}ms.", claimHold.toMillis());
         }
 
-        return new ExpiringResourceClaim(etcd, poolSize, claimHold, acquisitionTimeout);
+        return new ExpiringResourceClaim(etcd, maxGeneratorCount, clusterIds, claimHold, acquisitionTimeout);
     }
 }