You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@twill.apache.org by ch...@apache.org on 2013/11/21 22:54:25 UTC
[02/15] Initial import commit.
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/RewatchOnExpireWatcher.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/RewatchOnExpireWatcher.java b/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/RewatchOnExpireWatcher.java
new file mode 100644
index 0000000..181ca2b
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/RewatchOnExpireWatcher.java
@@ -0,0 +1,207 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.internal.zookeeper;
+
+import org.apache.twill.zookeeper.NodeChildren;
+import org.apache.twill.zookeeper.NodeData;
+import org.apache.twill.zookeeper.ZKClient;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.atomic.AtomicMarkableReference;
+
+/**
+ * A wrapper for {@link Watcher} that will re-set the watch automatically until it is successful.
+ */
+final class RewatchOnExpireWatcher implements Watcher {
+
+ private static final Logger LOG = LoggerFactory.getLogger(RewatchOnExpireWatcher.class);
+
+ enum ActionType {
+ EXISTS,
+ CHILDREN,
+ DATA
+ }
+
+ private final ZKClient client;
+ private final ActionType actionType;
+ private final String path;
+ private final Watcher delegate;
+ private final AtomicMarkableReference<Object> lastResult;
+
+ RewatchOnExpireWatcher(ZKClient client, ActionType actionType, String path, Watcher delegate) {
+ this.client = client;
+ this.actionType = actionType;
+ this.path = path;
+ this.delegate = delegate;
+ this.lastResult = new AtomicMarkableReference<Object>(null, false);
+ }
+
+ /**
+ * Sets the result from the operation that causes this watcher to be set.
+ */
+ void setLastResult(Object result) {
+ lastResult.compareAndSet(null, result, false, true);
+ }
+
+ @Override
+ public void process(WatchedEvent event) {
+ if (delegate != null && event.getType() != Event.EventType.None) {
+ try {
+ delegate.process(event);
+ } catch (Throwable t) {
+ LOG.error("Watcher throws exception.", t);
+ }
+ }
+
+ if (event.getState() != Event.KeeperState.Expired) {
+ return;
+ }
+ switch (actionType) {
+ case EXISTS:
+ exists();
+ break;
+ case CHILDREN:
+ children();
+ break;
+ case DATA:
+ data();
+ break;
+ }
+ }
+
+ private void exists() {
+ Futures.addCallback(client.exists(path, this), new FutureCallback<Stat>() {
+ @Override
+ public void onSuccess(Stat stat) {
+ // Since we know all callbacks and watcher are triggered from single event thread, there is no race condition.
+ Object oldResult = lastResult.getReference();
+ lastResult.compareAndSet(oldResult, null, true, false);
+
+ if (stat != oldResult && (stat == null || !stat.equals(oldResult))) {
+ if (stat == null) {
+ // previous stat is not null, means node deleted
+ process(new WatchedEvent(Event.EventType.NodeDeleted, Event.KeeperState.SyncConnected, path));
+ } else if (oldResult == null) {
+ // previous stat is null, means node created
+ process(new WatchedEvent(Event.EventType.NodeCreated, Event.KeeperState.SyncConnected, path));
+ } else {
+ // Otherwise, something changed on the node
+ process(new WatchedEvent(Event.EventType.NodeDataChanged, Event.KeeperState.SyncConnected, path));
+ }
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ if (RetryUtils.canRetry(t)) {
+ exists();
+ } else {
+ lastResult.set(null, false);
+ LOG.error("Fail to re-set watch on exists for path " + path, t);
+ }
+ }
+ });
+ }
+
+ private void children() {
+ Futures.addCallback(client.getChildren(path, this), new FutureCallback<NodeChildren>() {
+ @Override
+ public void onSuccess(NodeChildren result) {
+ Object oldResult = lastResult.getReference();
+ lastResult.compareAndSet(oldResult, null, true, false);
+
+ if (result.equals(oldResult)) {
+ return;
+ }
+
+ if (!(oldResult instanceof NodeChildren)) {
+ // Something very wrong
+ LOG.error("The same watcher has been used for different event type.");
+ return;
+ }
+
+ NodeChildren oldNodeChildren = (NodeChildren) oldResult;
+ if (!result.getChildren().equals(oldNodeChildren.getChildren())) {
+ process(new WatchedEvent(Event.EventType.NodeChildrenChanged, Event.KeeperState.SyncConnected, path));
+ } else {
+ process(new WatchedEvent(Event.EventType.NodeDataChanged, Event.KeeperState.SyncConnected, path));
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ if (RetryUtils.canRetry(t)) {
+ children();
+ return;
+ }
+
+ lastResult.set(null, false);
+ if (t instanceof KeeperException) {
+ KeeperException.Code code = ((KeeperException) t).code();
+ if (code == KeeperException.Code.NONODE) {
+ // Node deleted
+ process(new WatchedEvent(Event.EventType.NodeDeleted, Event.KeeperState.SyncConnected, path));
+ return;
+ }
+ }
+ LOG.error("Fail to re-set watch on getChildren for path " + path, t);
+ }
+ });
+ }
+
+ private void data() {
+ Futures.addCallback(client.getData(path, this), new FutureCallback<NodeData>() {
+ @Override
+ public void onSuccess(NodeData result) {
+ Object oldResult = lastResult.getReference();
+ lastResult.compareAndSet(oldResult, null, true, false);
+
+ if (!result.equals(oldResult)) {
+ // Whenever something changed, treated it as data changed.
+ process(new WatchedEvent(Event.EventType.NodeDataChanged, Event.KeeperState.SyncConnected, path));
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ if (RetryUtils.canRetry(t)) {
+ data();
+ return;
+ }
+
+ lastResult.set(null, false);
+ if (t instanceof KeeperException) {
+ KeeperException.Code code = ((KeeperException) t).code();
+ if (code == KeeperException.Code.NONODE) {
+ // Node deleted
+ process(new WatchedEvent(Event.EventType.NodeDeleted, Event.KeeperState.SyncConnected, path));
+ return;
+ }
+ }
+ LOG.error("Fail to re-set watch on getData for path " + path, t);
+ }
+ });
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/RewatchOnExpireZKClient.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/RewatchOnExpireZKClient.java b/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/RewatchOnExpireZKClient.java
new file mode 100644
index 0000000..402c153
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/RewatchOnExpireZKClient.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.internal.zookeeper;
+
+import org.apache.twill.internal.zookeeper.RewatchOnExpireWatcher.ActionType;
+import org.apache.twill.zookeeper.ForwardingZKClient;
+import org.apache.twill.zookeeper.NodeChildren;
+import org.apache.twill.zookeeper.NodeData;
+import org.apache.twill.zookeeper.OperationFuture;
+import org.apache.twill.zookeeper.ZKClient;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+
+/**
+ * A {@link ZKClient} that will rewatch automatically when session expired and reconnect.
+ * The rewatch logic is mainly done in {@link RewatchOnExpireWatcher}.
+ */
+public final class RewatchOnExpireZKClient extends ForwardingZKClient {
+
+ public RewatchOnExpireZKClient(ZKClient delegate) {
+ super(delegate);
+ }
+
+ @Override
+ public OperationFuture<Stat> exists(String path, Watcher watcher) {
+ final RewatchOnExpireWatcher wrappedWatcher = new RewatchOnExpireWatcher(this, ActionType.EXISTS, path, watcher);
+ OperationFuture<Stat> result = super.exists(path, wrappedWatcher);
+ Futures.addCallback(result, new FutureCallback<Stat>() {
+ @Override
+ public void onSuccess(Stat result) {
+ wrappedWatcher.setLastResult(result);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ // No-op
+ }
+ });
+ return result;
+ }
+
+ @Override
+ public OperationFuture<NodeChildren> getChildren(String path, Watcher watcher) {
+ final RewatchOnExpireWatcher wrappedWatcher = new RewatchOnExpireWatcher(this, ActionType.CHILDREN, path, watcher);
+ OperationFuture<NodeChildren> result = super.getChildren(path, wrappedWatcher);
+ Futures.addCallback(result, new FutureCallback<NodeChildren>() {
+ @Override
+ public void onSuccess(NodeChildren result) {
+ wrappedWatcher.setLastResult(result);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ // No-op
+ }
+ });
+ return result;
+ }
+
+ @Override
+ public OperationFuture<NodeData> getData(String path, Watcher watcher) {
+ final RewatchOnExpireWatcher wrappedWatcher = new RewatchOnExpireWatcher(this, ActionType.DATA, path, watcher);
+ OperationFuture<NodeData> result = super.getData(path, wrappedWatcher);
+ Futures.addCallback(result, new FutureCallback<NodeData>() {
+ @Override
+ public void onSuccess(NodeData result) {
+ wrappedWatcher.setLastResult(result);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ // No-op
+ }
+ });
+ return result;
+
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/SettableOperationFuture.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/SettableOperationFuture.java b/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/SettableOperationFuture.java
new file mode 100644
index 0000000..7544e56
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/SettableOperationFuture.java
@@ -0,0 +1,68 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.internal.zookeeper;
+
+import org.apache.twill.zookeeper.OperationFuture;
+import com.google.common.util.concurrent.AbstractFuture;
+
+import javax.annotation.Nullable;
+import java.util.concurrent.Executor;
+
+/**
+ * An implementation for {@link OperationFuture} that allows setting result directly.
+ * Also, all listener callback will be fired from the given executor.
+ */
+public final class SettableOperationFuture<V> extends AbstractFuture<V> implements OperationFuture<V> {
+
+ private final String requestPath;
+ private final Executor executor;
+
+ public static <V> SettableOperationFuture<V> create(String path, Executor executor) {
+ return new SettableOperationFuture<V>(path, executor);
+ }
+
+ private SettableOperationFuture(String requestPath, Executor executor) {
+ this.requestPath = requestPath;
+ this.executor = executor;
+ }
+
+ @Override
+ public String getRequestPath() {
+ return requestPath;
+ }
+
+ @Override
+ public void addListener(final Runnable listener, final Executor exec) {
+ super.addListener(new Runnable() {
+ @Override
+ public void run() {
+ exec.execute(listener);
+ }
+ }, executor);
+ }
+
+ @Override
+ public boolean setException(Throwable throwable) {
+ return super.setException(throwable);
+ }
+
+ @Override
+ public boolean set(@Nullable V value) {
+ return super.set(value);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/package-info.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/package-info.java b/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/package-info.java
new file mode 100644
index 0000000..d2afa11
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/internal/zookeeper/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * Internal classes for zookeeper.
+ */
+package org.apache.twill.internal.zookeeper;
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/ForwardingZKClient.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/ForwardingZKClient.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/ForwardingZKClient.java
new file mode 100644
index 0000000..3f3003d
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/ForwardingZKClient.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+
+import javax.annotation.Nullable;
+
+/**
+ *
+ */
+public abstract class ForwardingZKClient implements ZKClient {
+
+ private final ZKClient delegate;
+
+ protected ForwardingZKClient(ZKClient delegate) {
+ this.delegate = delegate;
+ }
+
+ public final ZKClient getDelegate() {
+ return delegate;
+ }
+
+ @Override
+ public Long getSessionId() {
+ return delegate.getSessionId();
+ }
+
+ @Override
+ public String getConnectString() {
+ return delegate.getConnectString();
+ }
+
+ @Override
+ public void addConnectionWatcher(Watcher watcher) {
+ delegate.addConnectionWatcher(watcher);
+ }
+
+ @Override
+ public OperationFuture<String> create(String path, @Nullable byte[] data, CreateMode createMode) {
+ return create(path, data, createMode, true);
+ }
+
+ @Override
+ public OperationFuture<String> create(String path, @Nullable byte[] data, CreateMode createMode,
+ boolean createParent) {
+ return delegate.create(path, data, createMode, createParent);
+ }
+
+ @Override
+ public OperationFuture<Stat> exists(String path) {
+ return exists(path, null);
+ }
+
+ @Override
+ public OperationFuture<Stat> exists(String path, @Nullable Watcher watcher) {
+ return delegate.exists(path, watcher);
+ }
+
+ @Override
+ public OperationFuture<NodeChildren> getChildren(String path) {
+ return getChildren(path, null);
+ }
+
+ @Override
+ public OperationFuture<NodeChildren> getChildren(String path, @Nullable Watcher watcher) {
+ return delegate.getChildren(path, watcher);
+ }
+
+ @Override
+ public OperationFuture<NodeData> getData(String path) {
+ return getData(path, null);
+ }
+
+ @Override
+ public OperationFuture<NodeData> getData(String path, @Nullable Watcher watcher) {
+ return delegate.getData(path, watcher);
+ }
+
+ @Override
+ public OperationFuture<Stat> setData(String path, byte[] data) {
+ return setData(path, data, -1);
+ }
+
+ @Override
+ public OperationFuture<Stat> setData(String dataPath, byte[] data, int version) {
+ return delegate.setData(dataPath, data, version);
+ }
+
+ @Override
+ public OperationFuture<String> delete(String path) {
+ return delete(path, -1);
+ }
+
+ @Override
+ public OperationFuture<String> delete(String deletePath, int version) {
+ return delegate.delete(deletePath, version);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/ForwardingZKClientService.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/ForwardingZKClientService.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/ForwardingZKClientService.java
new file mode 100644
index 0000000..10391b2
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/ForwardingZKClientService.java
@@ -0,0 +1,78 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import com.google.common.base.Supplier;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import org.apache.zookeeper.ZooKeeper;
+
+import java.util.concurrent.Executor;
+
+/**
+ *
+ */
+public abstract class ForwardingZKClientService extends ForwardingZKClient implements ZKClientService {
+
+ private final ZKClientService delegate;
+
+ protected ForwardingZKClientService(ZKClientService delegate) {
+ super(delegate);
+ this.delegate = delegate;
+ }
+
+ @Override
+ public Supplier<ZooKeeper> getZooKeeperSupplier() {
+ return delegate.getZooKeeperSupplier();
+ }
+
+ @Override
+ public ListenableFuture<State> start() {
+ return delegate.start();
+ }
+
+ @Override
+ public State startAndWait() {
+ return Futures.getUnchecked(start());
+ }
+
+ @Override
+ public boolean isRunning() {
+ return delegate.isRunning();
+ }
+
+ @Override
+ public State state() {
+ return delegate.state();
+ }
+
+ @Override
+ public ListenableFuture<State> stop() {
+ return delegate.stop();
+ }
+
+ @Override
+ public State stopAndWait() {
+ return Futures.getUnchecked(stop());
+ }
+
+ @Override
+ public void addListener(Listener listener, Executor executor) {
+ delegate.addListener(listener, executor);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/NodeChildren.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/NodeChildren.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/NodeChildren.java
new file mode 100644
index 0000000..b432c01
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/NodeChildren.java
@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.zookeeper.data.Stat;
+
+import java.util.List;
+
+/**
+ * Represents result of call to {@link ZKClientService#getChildren(String, org.apache.zookeeper.Watcher)} method.
+ */
+public interface NodeChildren {
+
+ /**
+ * @return The {@link Stat} of the node.
+ */
+ Stat getStat();
+
+ /**
+ * @return List of children node names.
+ */
+ List<String> getChildren();
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/NodeData.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/NodeData.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/NodeData.java
new file mode 100644
index 0000000..ac15957
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/NodeData.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.zookeeper.data.Stat;
+
+import javax.annotation.Nullable;
+
+/**
+ * Represents result of call to {@link ZKClientService#getData(String, org.apache.zookeeper.Watcher)}.
+ */
+public interface NodeData {
+
+ /**
+ * @return The {@link Stat} of the node.
+ */
+ Stat getStat();
+
+ /**
+ * @return Data stored in the node, or {@code null} if there is no data.
+ */
+ @Nullable
+ byte[] getData();
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/OperationFuture.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/OperationFuture.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/OperationFuture.java
new file mode 100644
index 0000000..fafaa7a
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/OperationFuture.java
@@ -0,0 +1,33 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import com.google.common.util.concurrent.ListenableFuture;
+
+/**
+ * A {@link ListenableFuture} that also provides the requested path for a operation.
+ *
+ * @param <V> The result type returned by this Future's {@link #get()} method.
+ */
+public interface OperationFuture<V> extends ListenableFuture<V> {
+
+ /**
+ * @return The path being requested for the ZooKeeper operation.
+ */
+ String getRequestPath();
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/RetryStrategies.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/RetryStrategies.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/RetryStrategies.java
new file mode 100644
index 0000000..56474b7
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/RetryStrategies.java
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import com.google.common.base.Preconditions;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Factory for creating common {@link RetryStrategy} implementation.
+ */
+public final class RetryStrategies {
+
+ /**
+ * @return A {@link RetryStrategy} that doesn't do any retry.
+ */
+ public static RetryStrategy noRetry() {
+ return new RetryStrategy() {
+ @Override
+ public long nextRetry(int failureCount, long startTime, OperationType type, String path) {
+ return -1;
+ }
+ };
+ }
+
+ /**
+ * Creates a {@link RetryStrategy} that retries maximum given number of times, with the actual
+ * delay behavior delegated to another {@link RetryStrategy}.
+ * @param limit Maximum number of retries allowed.
+ * @param strategy When failure count is less than or equal to the limit, this strategy will be called.
+ * @return A {@link RetryStrategy}.
+ */
+ public static RetryStrategy limit(final int limit, final RetryStrategy strategy) {
+ Preconditions.checkArgument(limit >= 0, "limit must be >= 0");
+ return new RetryStrategy() {
+ @Override
+ public long nextRetry(int failureCount, long startTime, OperationType type, String path) {
+ return (failureCount <= limit) ? strategy.nextRetry(failureCount, startTime, type, path) : -1L;
+ }
+ };
+ }
+
+ /**
+ * Creates a {@link RetryStrategy} that imposes a fix delay between each retries.
+ * @param delay delay time
+ * @param delayUnit {@link TimeUnit} for the delay.
+ * @return A {@link RetryStrategy}.
+ */
+ public static RetryStrategy fixDelay(final long delay, final TimeUnit delayUnit) {
+ Preconditions.checkArgument(delay >= 0, "delay must be >= 0");
+ return new RetryStrategy() {
+ @Override
+ public long nextRetry(int failureCount, long startTime, OperationType type, String path) {
+ return TimeUnit.MILLISECONDS.convert(delay, delayUnit);
+ }
+ };
+ }
+
+ /**
+ * Creates a {@link RetryStrategy} that will increase delay exponentially between each retries.
+ * @param baseDelay delay to start with.
+ * @param maxDelay cap of the delay.
+ * @param delayUnit {@link TimeUnit} for the delays.
+ * @return A {@link RetryStrategy}.
+ */
+ public static RetryStrategy exponentialDelay(final long baseDelay, final long maxDelay, final TimeUnit delayUnit) {
+ Preconditions.checkArgument(baseDelay >= 0, "base delay must be >= 0");
+ Preconditions.checkArgument(maxDelay >= 0, "max delay must be >= 0");
+ return new RetryStrategy() {
+ @Override
+ public long nextRetry(int failureCount, long startTime, OperationType type, String path) {
+ long power = failureCount > Long.SIZE ? Long.MAX_VALUE : (1L << (failureCount - 1));
+ long delay = Math.min(baseDelay * power, maxDelay);
+ delay = delay < 0 ? maxDelay : delay;
+ return TimeUnit.MILLISECONDS.convert(delay, delayUnit);
+ }
+ };
+ }
+
+ /**
+ * Creates a {@link RetryStrategy} that will retry until maximum amount of time has been passed since the request,
+ * with the actual delay behavior delegated to another {@link RetryStrategy}.
+ * @param maxElapseTime Maximum amount of time until giving up retry.
+ * @param timeUnit {@link TimeUnit} for the max elapse time.
+ * @param strategy When time elapsed is less than or equal to the limit, this strategy will be called.
+ * @return A {@link RetryStrategy}.
+ */
+ public static RetryStrategy timeLimit(long maxElapseTime, TimeUnit timeUnit, final RetryStrategy strategy) {
+ Preconditions.checkArgument(maxElapseTime >= 0, "max elapse time must be >= 0");
+ final long maxElapseMs = TimeUnit.MILLISECONDS.convert(maxElapseTime, timeUnit);
+ return new RetryStrategy() {
+ @Override
+ public long nextRetry(int failureCount, long startTime, OperationType type, String path) {
+ long elapseTime = System.currentTimeMillis() - startTime;
+ return elapseTime <= maxElapseMs ? strategy.nextRetry(failureCount, startTime, type, path) : -1L;
+ }
+ };
+ }
+
+ private RetryStrategies() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/RetryStrategy.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/RetryStrategy.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/RetryStrategy.java
new file mode 100644
index 0000000..3301e8a
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/RetryStrategy.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+/**
+ * Provides strategy to use for operation retries.
+ */
+public interface RetryStrategy {
+
+ /**
+ * Defines ZooKeeper operation type that triggers retry.
+ */
+ enum OperationType {
+ CREATE,
+ EXISTS,
+ GET_CHILDREN,
+ GET_DATA,
+ SET_DATA,
+ DELETE
+ }
+
+ /**
+ * Returns the number of milliseconds to wait before retrying the operation.
+ *
+ * @param failureCount Number of times that the request has been failed.
+ * @param startTime Timestamp in milliseconds that the request starts.
+ * @param type Type of operation tried to perform.
+ * @param path The path that the operation is acting on.
+ * @return Number of milliseconds to wait before retrying the operation. Returning {@code 0} means
+ * retry it immediately, while negative means abort the operation.
+ */
+ long nextRetry(int failureCount, long startTime, OperationType type, String path);
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClient.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClient.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClient.java
new file mode 100644
index 0000000..d60182e
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClient.java
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+
+import javax.annotation.Nullable;
+
+/**
+ * A ZooKeeper client that provides asynchronous zookeeper operations.
+ */
+public interface ZKClient {
+
+ /**
+ * Returns the current Zookeeper session ID of this client.
+ * If this ZKClient is not connected, {@code null} is returned.
+ */
+ Long getSessionId();
+
+ /**
+ * Returns the connection string used for connecting to Zookeeper.
+ */
+ String getConnectString();
+
+ /**
+ * Adds a {@link Watcher} that will be called whenever connection state change.
+ * @param watcher The watcher to set.
+ */
+ void addConnectionWatcher(Watcher watcher);
+
+ /**
+ * Same as calling
+ * {@link #create(String, byte[], org.apache.zookeeper.CreateMode, boolean) create(path, data, createMode, true)}.
+ *
+ * @see #create(String, byte[], org.apache.zookeeper.CreateMode, boolean)
+ */
+ OperationFuture<String> create(String path, @Nullable byte[] data, CreateMode createMode);
+
+ /**
+ * Creates a path in zookeeper, with given data and create mode.
+ *
+ * @param path Path to be created
+ * @param data Data to be stored in the node, or {@code null} if no data to store.
+ * @param createMode The {@link org.apache.zookeeper.CreateMode} for the node.
+ * @param createParent If {@code true} and parent nodes are missing, it will create all parent nodes as normal
+ * persistent node before creating the request node.
+ * @return A {@link OperationFuture} that will be completed when the
+ * creation is done. If there is error during creation, it will be reflected as error in the future.
+ */
+ OperationFuture<String> create(String path, @Nullable byte[] data, CreateMode createMode, boolean createParent);
+
+ /**
+ * Checks if the path exists. Same as calling
+ * {@link #exists(String, org.apache.zookeeper.Watcher) exists(path, null)}.
+ *
+ * @see #exists(String, org.apache.zookeeper.Watcher)
+ */
+ OperationFuture<Stat> exists(String path);
+
+ /**
+ * Checks if the given path exists and leave a watcher on the node for watching creation/deletion/data changes
+ * on the node.
+ *
+ * @param path The path to check for existence.
+ * @param watcher Watcher for watching changes, or {@code null} if no watcher to set.
+ * @return A {@link OperationFuture} that will be completed when the exists check is done. If the path
+ * does exists, the node {@link Stat} is set into the future. If the path doesn't exists,
+ * a {@code null} value is set into the future.
+ */
+ OperationFuture<Stat> exists(String path, @Nullable Watcher watcher);
+
+ /**
+ * Gets the list of children nodes under the given path. Same as calling
+ * {@link #getChildren(String, org.apache.zookeeper.Watcher) getChildren(path, null)}.
+ *
+ * @see #getChildren(String, org.apache.zookeeper.Watcher)
+ */
+ OperationFuture<NodeChildren> getChildren(String path);
+
+ /**
+ * Gets the list of children nodes under the given path and leave a watcher on the node for watching node
+ * deletion and children nodes creation/deletion.
+ *
+ * @param path The path to fetch for children nodes
+ * @param watcher Watcher for watching changes, or {@code null} if no watcher to set.
+ * @return A {@link OperationFuture} that will be completed when the getChildren call is done, with the result
+ * given as {@link NodeChildren}. If there is error, it will be reflected as error in the future.
+ */
+ OperationFuture<NodeChildren> getChildren(String path, @Nullable Watcher watcher);
+
+ /**
+ * Gets the data stored in the given path. Same as calling
+ * {@link #getData(String, org.apache.zookeeper.Watcher) getData(path, null)}.
+ */
+ OperationFuture<NodeData> getData(String path);
+
+ /**
+ * Gets the data stored in the given path and leave a watcher on the node for watching deletion/data changes on
+ * the node.
+ *
+ * @param path The path to get data from.
+ * @param watcher Watcher for watching changes, or {@code null} if no watcher to set.
+ * @return A {@link OperationFuture} that will be completed when the getData call is done, with the result
+ * given as {@link NodeData}. If there is error, it will be reflected as error in the future.
+ */
+ OperationFuture<NodeData> getData(String path, @Nullable Watcher watcher);
+
+ /**
+ * Sets the data for the given path without matching version. Same as calling
+ * {@link #setData(String, byte[], int) setData(path, data, -1)}.
+ */
+ OperationFuture<Stat> setData(String path, byte[] data);
+
+ /**
+ * Sets the data for the given path that match the given version. If the version given is {@code -1}, it matches
+ * any version.
+ *
+ * @param dataPath The path to set data to.
+ * @param data Data to be set.
+ * @param version Matching version.
+ * @return A {@link OperationFuture} that will be completed when the setData call is done, with node {@link Stat}
+ * given as the future result. If there is error, it will be reflected as error in the future.
+ */
+ OperationFuture<Stat> setData(String dataPath, byte[] data, int version);
+
+ /**
+ * Deletes the node of the given path without matching version. Same as calling
+ * {@link #delete(String, int) delete(path, -1)}.
+ *
+ * @see #delete(String, int)
+ */
+ OperationFuture<String> delete(String path);
+
+ /**
+ * Deletes the node of the given path that match the given version. If the version given is {@code -1}, it matches
+ * any version.
+ *
+ * @param deletePath The path to set data to.
+ * @param version Matching version.
+ * @return A {@link OperationFuture} that will be completed when the setData call is done, with node path
+ * given as the future result. If there is error, it will be reflected as error in the future.
+ */
+ OperationFuture<String> delete(String deletePath, int version);
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClientService.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClientService.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClientService.java
new file mode 100644
index 0000000..63f27fb
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClientService.java
@@ -0,0 +1,96 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.twill.internal.zookeeper.DefaultZKClientService;
+import com.google.common.base.Supplier;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+import com.google.common.util.concurrent.Service;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.ZooKeeper;
+import org.apache.zookeeper.data.ACL;
+
+/**
+ * A {@link ZKClient} that extends from {@link Service} to provide lifecycle management functions.
+ * The {@link #start()} method needed to be called before calling any other method on this interface.
+ * When the client is no longer needed, call {@link #stop()} to release any resources that it holds.
+ */
+public interface ZKClientService extends ZKClient, Service {
+
+ /**
+ * Returns a {@link Supplier} of {@link ZooKeeper} that gives the current {@link ZooKeeper} in use at the moment
+ * when {@link com.google.common.base.Supplier#get()} get called.
+ *
+ * @return A {@link Supplier Supplier<ZooKeeper>}
+ */
+ Supplier<ZooKeeper> getZooKeeperSupplier();
+
+ /**
+ * Builder for creating an implementation of {@link ZKClientService}.
+ * The default client timeout is 10000ms.
+ */
+ public static final class Builder {
+
+ private final String connectStr;
+ private int timeout = 10000;
+ private Watcher connectionWatcher;
+ private Multimap<String, ACL> acls = HashMultimap.create();
+
+ /**
+ * Creates a {@link Builder} with the given ZooKeeper connection string.
+ * @param connectStr The connection string.
+ * @return A new instance of Builder.
+ */
+ public static Builder of(String connectStr) {
+ return new Builder(connectStr);
+ }
+
+ /**
+ * Sets the client timeout to the give milliseconds.
+ * @param timeout timeout in milliseconds.
+ * @return This builder
+ */
+ public Builder setSessionTimeout(int timeout) {
+ this.timeout = timeout;
+ return this;
+ }
+
+ /**
+ * Sets a {@link Watcher} that will be called whenever connection state change.
+ * @param watcher The watcher to set.
+ * @return This builder.
+ */
+ public Builder setConnectionWatcher(Watcher watcher) {
+ this.connectionWatcher = watcher;
+ return this;
+ }
+
+ /**
+ * Creates an instance of {@link ZKClientService} with the settings of this builder.
+ * @return A new instance of {@link ZKClientService}.
+ */
+ public ZKClientService build() {
+ return new DefaultZKClientService(connectStr, timeout, connectionWatcher);
+ }
+
+ private Builder(String connectStr) {
+ this.connectStr = connectStr;
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClientServices.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClientServices.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClientServices.java
new file mode 100644
index 0000000..cc38c76
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClientServices.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+
+import javax.annotation.Nullable;
+
+/**
+ * Provides static factory method to create {@link ZKClientService} with modified behaviors.
+ */
+public final class ZKClientServices {
+
+ /**
+ * Creates a {@link ZKClientService} from the given {@link ZKClient} if the given {@link ZKClient} is an instance of
+ * {@link ZKClientService} or is a {@link ForwardingZKClient} that eventually trace back to a delegate of type
+ * {@link ZKClientService}. If such a {@link ZKClientService} instance is found, this method returns
+ * an instance by invoking {@link #delegate(ZKClient, ZKClientService)} with the given {@link ZKClient} and
+ * the {@link ZKClientService} found respectively.
+ *
+ * @param client The {@link ZKClient}.
+ * @return A {@link ZKClientService}.
+ * @throws IllegalArgumentException If no {@link ZKClientService} is found.
+ */
+ public static ZKClientService delegate(ZKClient client) {
+ ZKClient zkClient = client;
+ while (!(zkClient instanceof ZKClientService) && zkClient instanceof ForwardingZKClient) {
+ zkClient = ((ForwardingZKClient) zkClient).getDelegate();
+ }
+ if (zkClient instanceof ZKClientService) {
+ return delegate(client, (ZKClientService) zkClient);
+ }
+ throw new IllegalArgumentException("No ZKClientService found from the delegation hierarchy");
+ }
+
+ /**
+ * Creates a {@link ZKClientService} that for all {@link ZKClient} methods would be delegated to another
+ * {@link ZKClient}, while methods for {@link ZKClientService} would be delegated to another {@link ZKClientService},
+ * which the given {@link ZKClient} and {@link ZKClientService} could be different instances.
+ *
+ * @param client The {@link ZKClient} for delegation
+ * @param clientService The {@link ZKClientService} for delegation.
+ * @return A {@link ZKClientService}.
+ */
+ public static ZKClientService delegate(final ZKClient client, ZKClientService clientService) {
+ return new ForwardingZKClientService(clientService) {
+
+ @Override
+ public Long getSessionId() {
+ return client.getSessionId();
+ }
+
+ @Override
+ public String getConnectString() {
+ return client.getConnectString();
+ }
+
+ @Override
+ public void addConnectionWatcher(Watcher watcher) {
+ client.addConnectionWatcher(watcher);
+ }
+
+ @Override
+ public OperationFuture<String> create(String path, @Nullable byte[] data, CreateMode createMode) {
+ return client.create(path, data, createMode);
+ }
+
+ @Override
+ public OperationFuture<String> create(String path, @Nullable byte[] data, CreateMode createMode,
+ boolean createParent) {
+ return client.create(path, data, createMode, createParent);
+ }
+
+ @Override
+ public OperationFuture<Stat> exists(String path) {
+ return client.exists(path);
+ }
+
+ @Override
+ public OperationFuture<Stat> exists(String path, @Nullable Watcher watcher) {
+ return client.exists(path, watcher);
+ }
+
+ @Override
+ public OperationFuture<NodeChildren> getChildren(String path) {
+ return client.getChildren(path);
+ }
+
+ @Override
+ public OperationFuture<NodeChildren> getChildren(String path, @Nullable Watcher watcher) {
+ return client.getChildren(path, watcher);
+ }
+
+ @Override
+ public OperationFuture<NodeData> getData(String path) {
+ return client.getData(path);
+ }
+
+ @Override
+ public OperationFuture<NodeData> getData(String path, @Nullable Watcher watcher) {
+ return client.getData(path, watcher);
+ }
+
+ @Override
+ public OperationFuture<Stat> setData(String path, byte[] data) {
+ return client.setData(path, data);
+ }
+
+ @Override
+ public OperationFuture<Stat> setData(String dataPath, byte[] data, int version) {
+ return client.setData(dataPath, data, version);
+ }
+
+ @Override
+ public OperationFuture<String> delete(String path) {
+ return client.delete(path);
+ }
+
+ @Override
+ public OperationFuture<String> delete(String deletePath, int version) {
+ return client.delete(deletePath, version);
+ }
+ };
+ }
+
+ private ZKClientServices() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClients.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClients.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClients.java
new file mode 100644
index 0000000..f67c1bd
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKClients.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.twill.internal.zookeeper.FailureRetryZKClient;
+import org.apache.twill.internal.zookeeper.NamespaceZKClient;
+import org.apache.twill.internal.zookeeper.RewatchOnExpireZKClient;
+
+/**
+ *
+ */
+public final class ZKClients {
+
+ /**
+ * Creates a {@link ZKClient} that will perform auto re-watch on all existing watches
+ * when reconnection happens after session expiration. All {@link org.apache.zookeeper.Watcher Watchers}
+ * set through the returned {@link ZKClient} would not receive any connection events.
+ *
+ * @param client The {@link ZKClient} for operations delegation.
+ * @return A {@link ZKClient} that will do auto re-watch on all methods that accept a
+ * {@link org.apache.zookeeper.Watcher} upon session expiration.
+ */
+ public static ZKClient reWatchOnExpire(ZKClient client) {
+ return new RewatchOnExpireZKClient(client);
+ }
+
+ /**
+ * Creates a {@link ZKClient} that will retry interim failure (e.g. connection loss, session expiration)
+ * based on the given {@link RetryStrategy}.
+ *
+ * @param client The {@link ZKClient} for operations delegation.
+ * @param retryStrategy The {@link RetryStrategy} to be invoke when there is operation failure.
+ * @return A {@link ZKClient}.
+ */
+ public static ZKClient retryOnFailure(ZKClient client, RetryStrategy retryStrategy) {
+ return new FailureRetryZKClient(client, retryStrategy);
+ }
+
+
+ public static ZKClient namespace(ZKClient zkClient, String namespace) {
+ return new NamespaceZKClient(zkClient, namespace);
+ }
+
+ private ZKClients() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKOperations.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKOperations.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKOperations.java
new file mode 100644
index 0000000..6dcd1a7
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/ZKOperations.java
@@ -0,0 +1,355 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.twill.common.Cancellable;
+import org.apache.twill.common.Threads;
+import org.apache.twill.internal.zookeeper.SettableOperationFuture;
+import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+import org.apache.zookeeper.KeeperException;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.apache.zookeeper.data.Stat;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.List;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * Collection of helper methods for common operations that usually needed when interacting with ZooKeeper.
+ */
+public final class ZKOperations {
+
+ private static final Logger LOG = LoggerFactory.getLogger(ZKOperations.class);
+
+ /**
+ * Represents a ZK operation updates callback.
+ * @param <T> Type of updated data.
+ */
+ public interface Callback<T> {
+ void updated(T data);
+ }
+
+ /**
+ * Interface for defining callback method to receive node data updates.
+ */
+ public interface DataCallback extends Callback<NodeData> {
+ /**
+ * Invoked when data of the node changed.
+ * @param nodeData New data of the node, or {@code null} if the node has been deleted.
+ */
+ @Override
+ void updated(NodeData nodeData);
+ }
+
+ /**
+ * Interface for defining callback method to receive children nodes updates.
+ */
+ public interface ChildrenCallback extends Callback<NodeChildren> {
+ @Override
+ void updated(NodeChildren nodeChildren);
+ }
+
+ private interface Operation<T> {
+ ZKClient getZKClient();
+
+ OperationFuture<T> exec(String path, Watcher watcher);
+ }
+
+ /**
+ * Watch for data changes of the given path. The callback will be triggered whenever changes has been
+ * detected. Note that the callback won't see every single changes, as that's not the guarantee of ZooKeeper.
+ * If the node doesn't exists, it will watch for its creation then starts watching for data changes.
+ * When the node is deleted afterwards,
+ *
+ * @param zkClient The {@link ZKClient} for the operation
+ * @param path Path to watch
+ * @param callback Callback to be invoked when data changes is detected.
+ * @return A {@link Cancellable} to cancel the watch.
+ */
+ public static Cancellable watchData(final ZKClient zkClient, final String path, final DataCallback callback) {
+ final AtomicBoolean cancelled = new AtomicBoolean(false);
+ watchChanges(new Operation<NodeData>() {
+
+ @Override
+ public ZKClient getZKClient() {
+ return zkClient;
+ }
+
+ @Override
+ public OperationFuture<NodeData> exec(String path, Watcher watcher) {
+ return zkClient.getData(path, watcher);
+ }
+ }, path, callback, cancelled);
+
+ return new Cancellable() {
+ @Override
+ public void cancel() {
+ cancelled.set(true);
+ }
+ };
+ }
+
+ public static ListenableFuture<String> watchDeleted(final ZKClient zkClient, final String path) {
+ SettableFuture<String> completion = SettableFuture.create();
+ watchDeleted(zkClient, path, completion);
+ return completion;
+ }
+
+ public static void watchDeleted(final ZKClient zkClient, final String path,
+ final SettableFuture<String> completion) {
+
+ Futures.addCallback(zkClient.exists(path, new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (!completion.isDone()) {
+ if (event.getType() == Event.EventType.NodeDeleted) {
+ completion.set(path);
+ } else {
+ watchDeleted(zkClient, path, completion);
+ }
+ }
+ }
+ }), new FutureCallback<Stat>() {
+ @Override
+ public void onSuccess(Stat result) {
+ if (result == null) {
+ completion.set(path);
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ completion.setException(t);
+ }
+ });
+ }
+
+ public static Cancellable watchChildren(final ZKClient zkClient, String path, ChildrenCallback callback) {
+ final AtomicBoolean cancelled = new AtomicBoolean(false);
+ watchChanges(new Operation<NodeChildren>() {
+
+ @Override
+ public ZKClient getZKClient() {
+ return zkClient;
+ }
+
+ @Override
+ public OperationFuture<NodeChildren> exec(String path, Watcher watcher) {
+ return zkClient.getChildren(path, watcher);
+ }
+ }, path, callback, cancelled);
+
+ return new Cancellable() {
+ @Override
+ public void cancel() {
+ cancelled.set(true);
+ }
+ };
+ }
+
+ /**
+ * Returns a new {@link OperationFuture} that the result will be the same as the given future, except that when
+ * the source future is having an exception matching the giving exception type, the errorResult will be set
+ * in to the returned {@link OperationFuture}.
+ * @param future The source future.
+ * @param exceptionType Type of {@link KeeperException} to be ignored.
+ * @param errorResult Object to be set into the resulting future on a matching exception.
+ * @param <V> Type of the result.
+ * @return A new {@link OperationFuture}.
+ */
+ public static <V> OperationFuture<V> ignoreError(OperationFuture<V> future,
+ final Class<? extends KeeperException> exceptionType,
+ final V errorResult) {
+ final SettableOperationFuture<V> resultFuture = SettableOperationFuture.create(future.getRequestPath(),
+ Threads.SAME_THREAD_EXECUTOR);
+
+ Futures.addCallback(future, new FutureCallback<V>() {
+ @Override
+ public void onSuccess(V result) {
+ resultFuture.set(result);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ if (exceptionType.isAssignableFrom(t.getClass())) {
+ resultFuture.set(errorResult);
+ } else if (t instanceof CancellationException) {
+ resultFuture.cancel(true);
+ } else {
+ resultFuture.setException(t);
+ }
+ }
+ }, Threads.SAME_THREAD_EXECUTOR);
+
+ return resultFuture;
+ }
+
+ /**
+ * Deletes the given path recursively. The delete method will keep running until the given path is successfully
+ * removed, which means if there are new node created under the given path while deleting, they'll get deleted
+ * again. If there is {@link KeeperException} during the deletion other than
+ * {@link KeeperException.NotEmptyException} or {@link KeeperException.NoNodeException},
+ * the exception would be reflected in the result future and deletion process will stop,
+ * leaving the given path with intermediate state.
+ *
+ * @param path The path to delete.
+ * @return An {@link OperationFuture} that will be completed when the given path is deleted or bailed due to
+ * exception.
+ */
+ public static OperationFuture<String> recursiveDelete(final ZKClient zkClient, final String path) {
+ final SettableOperationFuture<String> resultFuture =
+ SettableOperationFuture.create(path, Threads.SAME_THREAD_EXECUTOR);
+
+ // Try to delete the given path.
+ Futures.addCallback(zkClient.delete(path), new FutureCallback<String>() {
+ private final FutureCallback<String> deleteCallback = this;
+
+ @Override
+ public void onSuccess(String result) {
+ // Path deleted successfully. Operation done.
+ resultFuture.set(result);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ // Failed to delete the given path
+ if (!(t instanceof KeeperException.NotEmptyException || t instanceof KeeperException.NoNodeException)) {
+ // For errors other than NotEmptyException, treat the operation as failed.
+ resultFuture.setException(t);
+ return;
+ }
+
+ // If failed because of NotEmptyException, get the list of children under the given path
+ Futures.addCallback(zkClient.getChildren(path), new FutureCallback<NodeChildren>() {
+
+ @Override
+ public void onSuccess(NodeChildren result) {
+ // Delete all children nodes recursively.
+ final List<OperationFuture<String>> deleteFutures = Lists.newLinkedList();
+ for (String child :result.getChildren()) {
+ deleteFutures.add(recursiveDelete(zkClient, path + "/" + child));
+ }
+
+ // When deletion of all children succeeded, delete the given path again.
+ Futures.successfulAsList(deleteFutures).addListener(new Runnable() {
+ @Override
+ public void run() {
+ for (OperationFuture<String> deleteFuture : deleteFutures) {
+ try {
+ // If any exception when deleting children, treat the operation as failed.
+ deleteFuture.get();
+ } catch (Exception e) {
+ resultFuture.setException(e.getCause());
+ }
+ }
+ Futures.addCallback(zkClient.delete(path), deleteCallback, Threads.SAME_THREAD_EXECUTOR);
+ }
+ }, Threads.SAME_THREAD_EXECUTOR);
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ // If failed to get list of children, treat the operation as failed.
+ resultFuture.setException(t);
+ }
+ }, Threads.SAME_THREAD_EXECUTOR);
+ }
+ }, Threads.SAME_THREAD_EXECUTOR);
+
+ return resultFuture;
+ }
+
+ /**
+ * Watch for the given path until it exists.
+ * @param zkClient The {@link ZKClient} to use.
+ * @param path A ZooKeeper path to watch for existent.
+ */
+ private static void watchExists(final ZKClient zkClient, final String path, final SettableFuture<String> completion) {
+ Futures.addCallback(zkClient.exists(path, new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (!completion.isDone()) {
+ watchExists(zkClient, path, completion);
+ }
+ }
+ }), new FutureCallback<Stat>() {
+ @Override
+ public void onSuccess(Stat result) {
+ if (result != null) {
+ completion.set(path);
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ completion.setException(t);
+ }
+ });
+ }
+
+ private static <T> void watchChanges(final Operation<T> operation, final String path,
+ final Callback<T> callback, final AtomicBoolean cancelled) {
+ Futures.addCallback(operation.exec(path, new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (!cancelled.get()) {
+ watchChanges(operation, path, callback, cancelled);
+ }
+ }
+ }), new FutureCallback<T>() {
+ @Override
+ public void onSuccess(T result) {
+ if (!cancelled.get()) {
+ callback.updated(result);
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ if (t instanceof KeeperException && ((KeeperException) t).code() == KeeperException.Code.NONODE) {
+ final SettableFuture<String> existCompletion = SettableFuture.create();
+ existCompletion.addListener(new Runnable() {
+ @Override
+ public void run() {
+ try {
+ if (!cancelled.get()) {
+ watchChanges(operation, existCompletion.get(), callback, cancelled);
+ }
+ } catch (Exception e) {
+ LOG.error("Failed to watch children for path " + path, e);
+ }
+ }
+ }, Threads.SAME_THREAD_EXECUTOR);
+ watchExists(operation.getZKClient(), path, existCompletion);
+ return;
+ }
+ LOG.error("Failed to watch data for path " + path + " " + t, t);
+ }
+ });
+ }
+
+ private ZKOperations() {
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/main/java/org/apache/twill/zookeeper/package-info.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/main/java/org/apache/twill/zookeeper/package-info.java b/zookeeper/src/main/java/org/apache/twill/zookeeper/package-info.java
new file mode 100644
index 0000000..e5bd237
--- /dev/null
+++ b/zookeeper/src/main/java/org/apache/twill/zookeeper/package-info.java
@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+/**
+ * This package provides functionality for ZooKeeper interactions.
+ */
+package org.apache.twill.zookeeper;
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/test/java/org/apache/twill/zookeeper/RetryStrategyTest.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/test/java/org/apache/twill/zookeeper/RetryStrategyTest.java b/zookeeper/src/test/java/org/apache/twill/zookeeper/RetryStrategyTest.java
new file mode 100644
index 0000000..601f0bd
--- /dev/null
+++ b/zookeeper/src/test/java/org/apache/twill/zookeeper/RetryStrategyTest.java
@@ -0,0 +1,94 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ *
+ */
+public class RetryStrategyTest {
+
+ @Test
+ public void testNoRetry() {
+ RetryStrategy strategy = RetryStrategies.noRetry();
+ long startTime = System.currentTimeMillis();
+ for (int i = 1; i <= 10; i++) {
+ Assert.assertEquals(-1L, strategy.nextRetry(i, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+ }
+
+ @Test
+ public void testLimit() {
+ RetryStrategy strategy = RetryStrategies.limit(10, RetryStrategies.fixDelay(1, TimeUnit.MILLISECONDS));
+ long startTime = System.currentTimeMillis();
+ for (int i = 1; i <= 10; i++) {
+ Assert.assertEquals(1L, strategy.nextRetry(i, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+ Assert.assertEquals(-1L, strategy.nextRetry(11, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+
+ @Test
+ public void testUnlimited() {
+ RetryStrategy strategy = RetryStrategies.fixDelay(1, TimeUnit.MILLISECONDS);
+ long startTime = System.currentTimeMillis();
+ for (int i = 1; i <= 10; i++) {
+ Assert.assertEquals(1L, strategy.nextRetry(i, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+ Assert.assertEquals(1L, strategy.nextRetry(100000, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+
+ @Test
+ public void testExponential() {
+ RetryStrategy strategy = RetryStrategies.exponentialDelay(1, 60000, TimeUnit.MILLISECONDS);
+ long startTime = System.currentTimeMillis();
+ for (int i = 1; i <= 16; i++) {
+ Assert.assertEquals(1L << (i - 1), strategy.nextRetry(i, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+ for (int i = 60; i <= 80; i++) {
+ Assert.assertEquals(60000, strategy.nextRetry(i, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+ }
+
+ @Test
+ public void testExponentialLimit() {
+ RetryStrategy strategy = RetryStrategies.limit(99,
+ RetryStrategies.exponentialDelay(1, 60000, TimeUnit.MILLISECONDS));
+ long startTime = System.currentTimeMillis();
+ for (int i = 1; i <= 16; i++) {
+ Assert.assertEquals(1L << (i - 1), strategy.nextRetry(i, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+ for (int i = 60; i <= 80; i++) {
+ Assert.assertEquals(60000, strategy.nextRetry(i, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+ Assert.assertEquals(-1L, strategy.nextRetry(100, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+
+ @Test
+ public void testTimeLimit() throws InterruptedException {
+ RetryStrategy strategy = RetryStrategies.timeLimit(1, TimeUnit.SECONDS,
+ RetryStrategies.fixDelay(1, TimeUnit.MILLISECONDS));
+ long startTime = System.currentTimeMillis();
+ Assert.assertEquals(1L, strategy.nextRetry(1, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ TimeUnit.MILLISECONDS.sleep(1100);
+ Assert.assertEquals(-1L, strategy.nextRetry(2, startTime, RetryStrategy.OperationType.CREATE, "/"));
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/test/java/org/apache/twill/zookeeper/ZKClientTest.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/test/java/org/apache/twill/zookeeper/ZKClientTest.java b/zookeeper/src/test/java/org/apache/twill/zookeeper/ZKClientTest.java
new file mode 100644
index 0000000..f1db74a
--- /dev/null
+++ b/zookeeper/src/test/java/org/apache/twill/zookeeper/ZKClientTest.java
@@ -0,0 +1,254 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.twill.internal.zookeeper.InMemoryZKServer;
+import org.apache.twill.internal.zookeeper.KillZKSession;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+import com.google.common.collect.Lists;
+import com.google.common.io.Files;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import org.apache.zookeeper.CreateMode;
+import org.apache.zookeeper.WatchedEvent;
+import org.apache.zookeeper.Watcher;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ *
+ */
+public class ZKClientTest {
+
+ @Test
+ public void testChroot() throws Exception {
+ InMemoryZKServer zkServer = InMemoryZKServer.builder().setTickTime(1000).build();
+ zkServer.startAndWait();
+
+ try {
+ ZKClientService client = ZKClientService.Builder.of(zkServer.getConnectionStr() + "/chroot").build();
+ client.startAndWait();
+ try {
+ List<OperationFuture<String>> futures = Lists.newArrayList();
+ futures.add(client.create("/test1/test2", null, CreateMode.PERSISTENT));
+ futures.add(client.create("/test1/test3", null, CreateMode.PERSISTENT));
+ Futures.successfulAsList(futures).get();
+
+ Assert.assertNotNull(client.exists("/test1/test2").get());
+ Assert.assertNotNull(client.exists("/test1/test3").get());
+
+ } finally {
+ client.stopAndWait();
+ }
+ } finally {
+ zkServer.stopAndWait();
+ }
+ }
+
+ @Test
+ public void testCreateParent() throws ExecutionException, InterruptedException {
+ InMemoryZKServer zkServer = InMemoryZKServer.builder().setTickTime(1000).build();
+ zkServer.startAndWait();
+
+ try {
+ ZKClientService client = ZKClientService.Builder.of(zkServer.getConnectionStr()).build();
+ client.startAndWait();
+
+ try {
+ String path = client.create("/test1/test2/test3/test4/test5",
+ "testing".getBytes(), CreateMode.PERSISTENT_SEQUENTIAL).get();
+ Assert.assertTrue(path.startsWith("/test1/test2/test3/test4/test5"));
+
+ String dataPath = "";
+ for (int i = 1; i <= 4; i++) {
+ dataPath = dataPath + "/test" + i;
+ Assert.assertNull(client.getData(dataPath).get().getData());
+ }
+ Assert.assertTrue(Arrays.equals("testing".getBytes(), client.getData(path).get().getData()));
+ } finally {
+ client.stopAndWait();
+ }
+ } finally {
+ zkServer.stopAndWait();
+ }
+ }
+
+ @Test
+ public void testGetChildren() throws ExecutionException, InterruptedException {
+ InMemoryZKServer zkServer = InMemoryZKServer.builder().setTickTime(1000).build();
+ zkServer.startAndWait();
+
+ try {
+ ZKClientService client = ZKClientService.Builder.of(zkServer.getConnectionStr()).build();
+ client.startAndWait();
+
+ try {
+ client.create("/test", null, CreateMode.PERSISTENT).get();
+ Assert.assertTrue(client.getChildren("/test").get().getChildren().isEmpty());
+
+ Futures.allAsList(ImmutableList.of(client.create("/test/c1", null, CreateMode.EPHEMERAL),
+ client.create("/test/c2", null, CreateMode.EPHEMERAL))).get();
+
+ NodeChildren nodeChildren = client.getChildren("/test").get();
+ Assert.assertEquals(2, nodeChildren.getChildren().size());
+
+ Assert.assertEquals(ImmutableSet.of("c1", "c2"), ImmutableSet.copyOf(nodeChildren.getChildren()));
+
+ } finally {
+ client.stopAndWait();
+ }
+ } finally {
+ zkServer.stopAndWait();
+ }
+ }
+
+ @Test
+ public void testSetData() throws ExecutionException, InterruptedException {
+ InMemoryZKServer zkServer = InMemoryZKServer.builder().setTickTime(1000).build();
+ zkServer.startAndWait();
+
+ try {
+ ZKClientService client = ZKClientService.Builder.of(zkServer.getConnectionStr()).build();
+ client.startAndWait();
+
+ client.create("/test", null, CreateMode.PERSISTENT).get();
+ Assert.assertNull(client.getData("/test").get().getData());
+
+ client.setData("/test", "testing".getBytes()).get();
+ Assert.assertTrue(Arrays.equals("testing".getBytes(), client.getData("/test").get().getData()));
+
+ } finally {
+ zkServer.stopAndWait();
+ }
+ }
+
+ @Test
+ public void testExpireRewatch() throws InterruptedException, IOException, ExecutionException {
+ InMemoryZKServer zkServer = InMemoryZKServer.builder().setTickTime(1000).build();
+ zkServer.startAndWait();
+
+ try {
+ final CountDownLatch expireReconnectLatch = new CountDownLatch(1);
+ final AtomicBoolean expired = new AtomicBoolean(false);
+ final ZKClientService client = ZKClientServices.delegate(ZKClients.reWatchOnExpire(
+ ZKClientService.Builder.of(zkServer.getConnectionStr())
+ .setSessionTimeout(2000)
+ .setConnectionWatcher(new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (event.getState() == Event.KeeperState.Expired) {
+ expired.set(true);
+ } else if (event.getState() == Event.KeeperState.SyncConnected && expired.compareAndSet(true, true)) {
+ expireReconnectLatch.countDown();
+ }
+ }
+ }).build()));
+ client.startAndWait();
+
+ try {
+ final BlockingQueue<Watcher.Event.EventType> events = new LinkedBlockingQueue<Watcher.Event.EventType>();
+ client.exists("/expireRewatch", new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ client.exists("/expireRewatch", this);
+ events.add(event.getType());
+ }
+ });
+
+ client.create("/expireRewatch", null, CreateMode.PERSISTENT);
+ Assert.assertEquals(Watcher.Event.EventType.NodeCreated, events.poll(2, TimeUnit.SECONDS));
+
+ KillZKSession.kill(client.getZooKeeperSupplier().get(), zkServer.getConnectionStr(), 1000);
+
+ Assert.assertTrue(expireReconnectLatch.await(5, TimeUnit.SECONDS));
+
+ client.delete("/expireRewatch");
+ Assert.assertEquals(Watcher.Event.EventType.NodeDeleted, events.poll(4, TimeUnit.SECONDS));
+ } finally {
+ client.stopAndWait();
+ }
+ } finally {
+ zkServer.stopAndWait();
+ }
+ }
+
+ @Test
+ public void testRetry() throws ExecutionException, InterruptedException, TimeoutException {
+ File dataDir = Files.createTempDir();
+ InMemoryZKServer zkServer = InMemoryZKServer.builder().setDataDir(dataDir).setTickTime(1000).build();
+ zkServer.startAndWait();
+ int port = zkServer.getLocalAddress().getPort();
+
+ final CountDownLatch disconnectLatch = new CountDownLatch(1);
+ ZKClientService client = ZKClientServices.delegate(ZKClients.retryOnFailure(
+ ZKClientService.Builder.of(zkServer.getConnectionStr()).setConnectionWatcher(new Watcher() {
+ @Override
+ public void process(WatchedEvent event) {
+ if (event.getState() == Event.KeeperState.Disconnected) {
+ disconnectLatch.countDown();
+ }
+ }
+ }).build(), RetryStrategies.fixDelay(0, TimeUnit.SECONDS)));
+ client.startAndWait();
+
+ zkServer.stopAndWait();
+
+ Assert.assertTrue(disconnectLatch.await(1, TimeUnit.SECONDS));
+
+ final CountDownLatch createLatch = new CountDownLatch(1);
+ Futures.addCallback(client.create("/testretry/test", null, CreateMode.PERSISTENT), new FutureCallback<String>() {
+ @Override
+ public void onSuccess(String result) {
+ createLatch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable t) {
+ t.printStackTrace(System.out);
+ }
+ });
+
+ TimeUnit.SECONDS.sleep(2);
+ zkServer = InMemoryZKServer.builder()
+ .setDataDir(dataDir)
+ .setAutoCleanDataDir(true)
+ .setPort(port)
+ .setTickTime(1000)
+ .build();
+ zkServer.startAndWait();
+
+ try {
+ Assert.assertTrue(createLatch.await(5, TimeUnit.SECONDS));
+ } finally {
+ zkServer.stopAndWait();
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-twill/blob/1925ffaf/zookeeper/src/test/java/org/apache/twill/zookeeper/ZKOperationsTest.java
----------------------------------------------------------------------
diff --git a/zookeeper/src/test/java/org/apache/twill/zookeeper/ZKOperationsTest.java b/zookeeper/src/test/java/org/apache/twill/zookeeper/ZKOperationsTest.java
new file mode 100644
index 0000000..9518d6e
--- /dev/null
+++ b/zookeeper/src/test/java/org/apache/twill/zookeeper/ZKOperationsTest.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.twill.zookeeper;
+
+import org.apache.twill.internal.zookeeper.InMemoryZKServer;
+import org.apache.zookeeper.CreateMode;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+/**
+ *
+ */
+public class ZKOperationsTest {
+
+ @Test
+ public void recursiveDelete() throws ExecutionException, InterruptedException, TimeoutException {
+ InMemoryZKServer zkServer = InMemoryZKServer.builder().setTickTime(1000).build();
+ zkServer.startAndWait();
+
+ try {
+ ZKClientService client = ZKClientService.Builder.of(zkServer.getConnectionStr()).build();
+ client.startAndWait();
+
+ try {
+ client.create("/test1/test10/test101", null, CreateMode.PERSISTENT).get();
+ client.create("/test1/test10/test102", null, CreateMode.PERSISTENT).get();
+ client.create("/test1/test10/test103", null, CreateMode.PERSISTENT).get();
+
+ client.create("/test1/test11/test111", null, CreateMode.PERSISTENT).get();
+ client.create("/test1/test11/test112", null, CreateMode.PERSISTENT).get();
+ client.create("/test1/test11/test113", null, CreateMode.PERSISTENT).get();
+
+ ZKOperations.recursiveDelete(client, "/test1").get(2, TimeUnit.SECONDS);
+
+ Assert.assertNull(client.exists("/test1").get(2, TimeUnit.SECONDS));
+
+ } finally {
+ client.stopAndWait();
+ }
+ } finally {
+ zkServer.stopAndWait();
+ }
+ }
+}