add defaultWaitForShutdownTimeoutMs
Project: http://git-wip-us.apache.org/repos/asf/curator/repo Commit: http://git-wip-us.apache.org/repos/asf/curator/commit/aec13472 Tree: http://git-wip-us.apache.org/repos/asf/curator/tree/aec13472 Diff: http://git-wip-us.apache.org/repos/asf/curator/diff/aec13472 Branch: refs/heads/master Commit: aec134724a500120a11542d43659a676562b03a1 Parents: fe2c7c4 Author: Enrico Olivelli <[email protected]> Authored: Sun Jun 24 18:05:11 2018 +0200 Committer: Enrico Olivelli <[email protected]> Committed: Sun Jun 24 18:05:11 2018 +0200 ---------------------------------------------------------------------- .../apache/curator/CuratorZookeeperClient.java | 34 ++++++++++++++++++-- .../framework/CuratorFrameworkFactory.java | 22 ++++++++++++- .../framework/imps/CuratorFrameworkImpl.java | 1 + 3 files changed, 53 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/curator/blob/aec13472/curator-client/src/main/java/org/apache/curator/CuratorZookeeperClient.java ---------------------------------------------------------------------- diff --git a/curator-client/src/main/java/org/apache/curator/CuratorZookeeperClient.java b/curator-client/src/main/java/org/apache/curator/CuratorZookeeperClient.java index dd3ae82..065743b 100644 --- a/curator-client/src/main/java/org/apache/curator/CuratorZookeeperClient.java +++ b/curator-client/src/main/java/org/apache/curator/CuratorZookeeperClient.java @@ -52,6 +52,7 @@ public class CuratorZookeeperClient implements Closeable private final ConnectionState state; private final AtomicReference<RetryPolicy> retryPolicy = new AtomicReference<RetryPolicy>(); private final int connectionTimeoutMs; + private final int defaultWaitForShutdownTimeoutMs; private final AtomicBoolean started = new AtomicBoolean(false); private final AtomicReference<TracerDriver> tracer = new AtomicReference<TracerDriver>(new DefaultTracerDriver()); private final ConnectionHandlingPolicy connectionHandlingPolicy; @@ -112,7 +113,28 @@ public class CuratorZookeeperClient implements Closeable * @param connectionHandlingPolicy connection handling policy - use one of the pre-defined policies or write your own * @since 3.0.0 */ - public CuratorZookeeperClient(ZookeeperFactory zookeeperFactory, EnsembleProvider ensembleProvider, int sessionTimeoutMs, int connectionTimeoutMs, Watcher watcher, RetryPolicy retryPolicy, boolean canBeReadOnly, ConnectionHandlingPolicy connectionHandlingPolicy) + public CuratorZookeeperClient(ZookeeperFactory zookeeperFactory, EnsembleProvider ensembleProvider, int sessionTimeoutMs, int connectionTimeoutMs, Watcher watcher, RetryPolicy retryPolicy, boolean canBeReadOnly, ConnectionHandlingPolicy connectionHandlingPolicy) { + this(zookeeperFactory, ensembleProvider, sessionTimeoutMs, connectionTimeoutMs, 0, + watcher, retryPolicy, canBeReadOnly, connectionHandlingPolicy); + } + /** + * @param zookeeperFactory factory for creating {@link ZooKeeper} instances + * @param ensembleProvider the ensemble provider + * @param sessionTimeoutMs session timeout + * @param connectionTimeoutMs connection timeout + * @param defaultWaitForShutdownTimeoutMs default timeout fo close operation + * @param watcher default watcher or null + * @param retryPolicy the retry policy to use + * @param canBeReadOnly if true, allow ZooKeeper client to enter + * read only mode in case of a network partition. See + * {@link ZooKeeper#ZooKeeper(String, int, Watcher, long, byte[], boolean)} + * for details + * @param connectionHandlingPolicy connection handling policy - use one of the pre-defined policies or write your own + * @since 4.0.2 + */ + public CuratorZookeeperClient(ZookeeperFactory zookeeperFactory, EnsembleProvider ensembleProvider, + int sessionTimeoutMs, int connectionTimeoutMs, int defaultWaitForShutdownTimeoutMs, Watcher watcher, + RetryPolicy retryPolicy, boolean canBeReadOnly, ConnectionHandlingPolicy connectionHandlingPolicy) { this.connectionHandlingPolicy = connectionHandlingPolicy; if ( sessionTimeoutMs < connectionTimeoutMs ) @@ -124,6 +146,7 @@ public class CuratorZookeeperClient implements Closeable ensembleProvider = Preconditions.checkNotNull(ensembleProvider, "ensembleProvider cannot be null"); this.connectionTimeoutMs = connectionTimeoutMs; + this.defaultWaitForShutdownTimeoutMs = defaultWaitForShutdownTimeoutMs; state = new ConnectionState(zookeeperFactory, ensembleProvider, sessionTimeoutMs, connectionTimeoutMs, watcher, tracer, canBeReadOnly, connectionHandlingPolicy); setRetryPolicy(retryPolicy); } @@ -215,10 +238,15 @@ public class CuratorZookeeperClient implements Closeable } /** - * Close the client + * Close the client. + * + * Same as {@link #close(int) } using the default timeout set at construction time. + * + * @see #close(int) */ + @Override public void close() { - close(0); + close(defaultWaitForShutdownTimeoutMs); } /** http://git-wip-us.apache.org/repos/asf/curator/blob/aec13472/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java index a617198..dad164e 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/CuratorFrameworkFactory.java @@ -49,6 +49,7 @@ import java.util.Arrays; import java.util.List; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; +import org.apache.curator.CuratorZookeeperClient; import static org.apache.curator.utils.Compatibility.isZK34; @@ -147,7 +148,7 @@ public class CuratorFrameworkFactory private ConnectionHandlingPolicy connectionHandlingPolicy = new StandardConnectionHandlingPolicy(); private SchemaSet schemaSet = SchemaSet.getDefaultSchemaSet(); private boolean zk34CompatibilityMode = isZK34(); - + private int defaultWaitForShutdownTimeoutMs = 0; /** * Apply the current values and build a new CuratorFramework * @@ -402,6 +403,20 @@ public class CuratorFrameworkFactory } /** + * Set a default timeout for {@link CuratorZookeeperClient#close() }. + * The default is 0, which means that this feature is disabled. + * + * @since 4.0.2 + * @param defaultWaitForShutdownTimeoutMs default timeout + * @return this + */ + public Builder defaultWaitForShutdownTimeoutMs(int defaultWaitForShutdownTimeoutMs) + { + this.defaultWaitForShutdownTimeoutMs = defaultWaitForShutdownTimeoutMs; + return this; + } + + /** * <p> * Change the connection handling policy. The default policy is {@link StandardConnectionHandlingPolicy}. * </p> @@ -494,6 +509,11 @@ public class CuratorFrameworkFactory return connectionTimeoutMs; } + public int getDefaultWaitForShutdownTimeoutMs() + { + return defaultWaitForShutdownTimeoutMs; + } + public int getMaxCloseWaitMs() { return maxCloseWaitMs; http://git-wip-us.apache.org/repos/asf/curator/blob/aec13472/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java ---------------------------------------------------------------------- diff --git a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java index 2bd5c7c..80a7b56 100644 --- a/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java +++ b/curator-framework/src/main/java/org/apache/curator/framework/imps/CuratorFrameworkImpl.java @@ -122,6 +122,7 @@ public class CuratorFrameworkImpl implements CuratorFramework builder.getEnsembleProvider(), builder.getSessionTimeoutMs(), builder.getConnectionTimeoutMs(), + builder.getDefaultWaitForShutdownTimeoutMs(), new Watcher() { @Override
