diff options
Diffstat (limited to 'zkfacade')
-rw-r--r-- | zkfacade/src/main/java/com/yahoo/vespa/curator/CuratorCompletionWaiter.java | 24 |
1 files changed, 13 insertions, 11 deletions
diff --git a/zkfacade/src/main/java/com/yahoo/vespa/curator/CuratorCompletionWaiter.java b/zkfacade/src/main/java/com/yahoo/vespa/curator/CuratorCompletionWaiter.java index 420b049cc50..7d918baaf54 100644 --- a/zkfacade/src/main/java/com/yahoo/vespa/curator/CuratorCompletionWaiter.java +++ b/zkfacade/src/main/java/com/yahoo/vespa/curator/CuratorCompletionWaiter.java @@ -9,6 +9,8 @@ import java.time.Instant; import java.util.List; import java.util.logging.Level; +import static com.yahoo.vespa.curator.Curator.CompletionWaiter; + /** * Implementation of a Barrier that handles the case where more than number of members can call synchronize. * Will wait for some time for all servers to do the operation, but will accept the majority of servers to have @@ -17,18 +19,18 @@ import java.util.logging.Level; * @author Vegard Havdal * @author Ulf Lilleengen */ -class CuratorCompletionWaiter implements Curator.CompletionWaiter { +class CuratorCompletionWaiter implements CompletionWaiter { private static final java.util.logging.Logger log = java.util.logging.Logger.getLogger(CuratorCompletionWaiter.class.getName()); private final Curator curator; - private final String barrierPath; - private final String myId; + private final Path barrierPath; + private final Path waiterNode; private final Clock clock; private final Duration waitForAll; - CuratorCompletionWaiter(Curator curator, String barrierPath, String myId, Clock clock, Duration waitForAll) { - this.myId = barrierPath + "/" + myId; + CuratorCompletionWaiter(Curator curator, Path barrierPath, String serverId, Clock clock, Duration waitForAll) { + this.waiterNode = barrierPath.append(serverId); this.curator = curator; this.barrierPath = barrierPath; this.clock = clock; @@ -61,7 +63,7 @@ class CuratorCompletionWaiter implements Curator.CompletionWaiter { List<String> respondents; do { - respondents = (curator.framework().getChildren().forPath(barrierPath)); + respondents = curator.framework().getChildren().forPath(barrierPath.getAbsolute()); if (log.isLoggable(Level.FINER)) { log.log(Level.FINER, respondents.size() + "/" + curator.zooKeeperEnsembleCount() + " responded: " + respondents + ", all participants: " + curator.zooKeeperEnsembleConnectionSpec()); @@ -103,7 +105,7 @@ class CuratorCompletionWaiter implements Curator.CompletionWaiter { @Override public void notifyCompletion() { try { - curator.framework().create().forPath(myId); + curator.framework().create().forPath(waiterNode.getAbsolute()); } catch (Exception e) { throw new RuntimeException(e); } @@ -114,15 +116,15 @@ class CuratorCompletionWaiter implements Curator.CompletionWaiter { return "'" + barrierPath + "', " + barrierMemberCount() + " members"; } - public static Curator.CompletionWaiter create(Curator curator, Path barrierPath, String id, Duration waitForAll) { - return new CuratorCompletionWaiter(curator, barrierPath.getAbsolute(), id, Clock.systemUTC(), waitForAll); + public static CompletionWaiter create(Curator curator, Path barrierPath, String id, Duration waitForAll) { + return new CuratorCompletionWaiter(curator, barrierPath, id, Clock.systemUTC(), waitForAll); } - public static Curator.CompletionWaiter createAndInitialize(Curator curator, Path waiterPath, String id, Duration waitForAll) { + public static CompletionWaiter createAndInitialize(Curator curator, Path waiterPath, String id, Duration waitForAll) { curator.delete(waiterPath); curator.createAtomically(waiterPath); - return new CuratorCompletionWaiter(curator, waiterPath.getAbsolute(), id, Clock.systemUTC(), waitForAll); + return new CuratorCompletionWaiter(curator, waiterPath, id, Clock.systemUTC(), waitForAll); } private int barrierMemberCount() { |