Details

    • Type: Improvement
    • Status: Closed
    • Priority: Minor
    • Resolution: Fixed
    • Affects Version/s: 1.2.0, 1.3.0
    • Fix Version/s: 1.2.0, 1.3.0
    • Component/s: Mesos
    • Labels:
      None

      Description

      Mesos uses dynamic class loading in order to load the ZooKeeperStateHandleStore and the CuratorFramework class. This can be replaced by a compile time dependency.

        Issue Links

          Activity

          Hide
          till.rohrmann Till Rohrmann added a comment -

          The ZooKeeperStateHandleStore cannot be loaded statically because it requires a CuratorFramework argument for instantiation. Since we relocate Curator in flink-runtime, the classes are not compatible. In order to fix the problem, we either have to introduce our own interface for CuratorFramework or refrain from relocating Curator in flink-runtime. For the time being, I'll only remove the unnecessary ZooKeeperUtils class from the flink-mesos module.

          Show
          till.rohrmann Till Rohrmann added a comment - The ZooKeeperStateHandleStore cannot be loaded statically because it requires a CuratorFramework argument for instantiation. Since we relocate Curator in flink-runtime , the classes are not compatible. In order to fix the problem, we either have to introduce our own interface for CuratorFramework or refrain from relocating Curator in flink-runtime . For the time being, I'll only remove the unnecessary ZooKeeperUtils class from the flink-mesos module.
          Hide
          till.rohrmann Till Rohrmann added a comment -

          Unfortunately, introducing a non relocated CuratorUtils class in flink-shaded-curator-recipes which can create a CuratorFramework does not solve the problem since flink-runtime classes such as ZooKeeperStateHandleStore would still require the relocated CuratorFramework as an argument. Thus, we can only complete this issue once FLINK-5513 has been completed.

          Show
          till.rohrmann Till Rohrmann added a comment - Unfortunately, introducing a non relocated CuratorUtils class in flink-shaded-curator-recipes which can create a CuratorFramework does not solve the problem since flink-runtime classes such as ZooKeeperStateHandleStore would still require the relocated CuratorFramework as an argument. Thus, we can only complete this issue once FLINK-5513 has been completed.
          Hide
          till.rohrmann Till Rohrmann added a comment -

          Strong requirement in order to complete this issue.

          Show
          till.rohrmann Till Rohrmann added a comment - Strong requirement in order to complete this issue.
          Hide
          till.rohrmann Till Rohrmann added a comment -

          As a temporary fix, we'll introduce a ZooKeeperStateHandleStoreFactory in flink-runtime in order to circumvent the problem of having to create and pass a CuratorFramework instance to it.

          Show
          till.rohrmann Till Rohrmann added a comment - As a temporary fix, we'll introduce a ZooKeeperStateHandleStoreFactory in flink-runtime in order to circumvent the problem of having to create and pass a CuratorFramework instance to it.
          Hide
          githubbot ASF GitHub Bot added a comment -

          GitHub user tillrohrmann opened a pull request:

          https://github.com/apache/flink/pull/3157

          FLINK-5508 [mesos] Introduce ZooKeeperUtilityFactory to create ZooKeeper utility classes

          This PR is based on #3155 and #3156.

          This commit adds utility classes to abstract the CuratorFramework dependency from ZooKeeper
          utility classes away. That way it is possible for modules outside of flink-runtime to use
          these utility classes without facing the problem of a relocated curator dependency.

          You can merge this pull request into a Git repository by running:

          $ git pull https://github.com/tillrohrmann/flink mesosRemoveDynamicBinding

          Alternatively you can review and apply these changes as the patch at:

          https://github.com/apache/flink/pull/3157.patch

          To close this pull request, make a commit to your master/trunk branch
          with (at least) the following in the commit message:

          This closes #3157


          commit e68688f31ead68851a0f768a7270d8bc1b5f9ac8
          Author: Till Rohrmann <trohrmann@apache.org>
          Date: 2017-01-16T13:01:10Z

          FLINK-5496 [mesos] Relocate Mesos Protobuf dependency to avoid version conflicts

          Only relocate Mesos Protobuf dependency in flink-mesos. This avoids problems with Mesos
          because Flink pulls in Protobuf 2.5.0 via Flakka.

          commit dd2f56568745ade036de4d0ee333e7e7fdb47400
          Author: Till Rohrmann <trohrmann@apache.org>
          Date: 2017-01-16T13:14:18Z

          FLINK-5495 [mesos] Provide executor to ZooKeeperMesosWorkerStore

          The ZooKeeperMesosWorkerStore instantiates a ZooKeeperStateHandleStore which requires an
          Executor instance. This executor is now given to the ZooKeeperMesosWorkerStore.

          commit 93630bac520b8fe4a223f2b724621b211293dad2
          Author: Till Rohrmann <trohrmann@apache.org>
          Date: 2017-01-18T14:06:12Z

          FLINK-5508 [mesos] Introduce ZooKeeperUtilityFactory to create ZooKeeper utility classes

          This commit adds utility classes to abstract the CuratorFramework dependency from ZooKeeper
          utility classes away. That way it is possible for modules outside of flink-runtime to use
          these utility classes without facing the problem of a relocated curator dependency.


          Show
          githubbot ASF GitHub Bot added a comment - GitHub user tillrohrmann opened a pull request: https://github.com/apache/flink/pull/3157 FLINK-5508 [mesos] Introduce ZooKeeperUtilityFactory to create ZooKeeper utility classes This PR is based on #3155 and #3156. This commit adds utility classes to abstract the CuratorFramework dependency from ZooKeeper utility classes away. That way it is possible for modules outside of flink-runtime to use these utility classes without facing the problem of a relocated curator dependency. You can merge this pull request into a Git repository by running: $ git pull https://github.com/tillrohrmann/flink mesosRemoveDynamicBinding Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3157.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3157 commit e68688f31ead68851a0f768a7270d8bc1b5f9ac8 Author: Till Rohrmann <trohrmann@apache.org> Date: 2017-01-16T13:01:10Z FLINK-5496 [mesos] Relocate Mesos Protobuf dependency to avoid version conflicts Only relocate Mesos Protobuf dependency in flink-mesos. This avoids problems with Mesos because Flink pulls in Protobuf 2.5.0 via Flakka. commit dd2f56568745ade036de4d0ee333e7e7fdb47400 Author: Till Rohrmann <trohrmann@apache.org> Date: 2017-01-16T13:14:18Z FLINK-5495 [mesos] Provide executor to ZooKeeperMesosWorkerStore The ZooKeeperMesosWorkerStore instantiates a ZooKeeperStateHandleStore which requires an Executor instance. This executor is now given to the ZooKeeperMesosWorkerStore. commit 93630bac520b8fe4a223f2b724621b211293dad2 Author: Till Rohrmann <trohrmann@apache.org> Date: 2017-01-18T14:06:12Z FLINK-5508 [mesos] Introduce ZooKeeperUtilityFactory to create ZooKeeper utility classes This commit adds utility classes to abstract the CuratorFramework dependency from ZooKeeper utility classes away. That way it is possible for modules outside of flink-runtime to use these utility classes without facing the problem of a relocated curator dependency.
          Hide
          githubbot ASF GitHub Bot added a comment -

          GitHub user tillrohrmann opened a pull request:

          https://github.com/apache/flink/pull/3158

          [backport] FLINK-5508 FLINK-5496 FLINK-5495 Fix Mesos HA problems

          This PR is a backport of #3155, #3156 and #3157 onto the `release-1.2` branch.

          `da3358a` fixes:
          The ZooKeeperMesosWorkerStore instantiates a ZooKeeperStateHandleStore which requires an
          Executor instance. This executor is now given to the ZooKeeperMesosWorkerStore.

          `e14fb2d` fixes:
          Only relocate Mesos Protobuf dependency in flink-mesos. This avoids problems with Mesos
          because Flink pulls in Protobuf 2.5.0 via Flakka.

          `b34a95c` fixes:
          This commit adds utility classes to abstract the CuratorFramework dependency from ZooKeeper
          utility classes away. That way it is possible for modules outside of flink-runtime to use
          these utility classes without facing the problem of a relocated curator dependency.

          You can merge this pull request into a Git repository by running:

          $ git pull https://github.com/tillrohrmann/flink backportCuratorShading

          Alternatively you can review and apply these changes as the patch at:

          https://github.com/apache/flink/pull/3158.patch

          To close this pull request, make a commit to your master/trunk branch
          with (at least) the following in the commit message:

          This closes #3158


          commit da3358ac5addac540910a9c492aece6435797b52
          Author: Till Rohrmann <trohrmann@apache.org>
          Date: 2017-01-16T13:14:18Z

          FLINK-5495 [mesos] Provide executor to ZooKeeperMesosWorkerStore

          The ZooKeeperMesosWorkerStore instantiates a ZooKeeperStateHandleStore which requires an
          Executor instance. This executor is now given to the ZooKeeperMesosWorkerStore.

          commit e14fb2de0414378fb32aa21e88cd2ac0ba956da5
          Author: Till Rohrmann <trohrmann@apache.org>
          Date: 2017-01-16T13:01:10Z

          FLINK-5496 [mesos] Relocate Mesos Protobuf dependency to avoid version conflicts

          Only relocate Mesos Protobuf dependency in flink-mesos. This avoids problems with Mesos
          because Flink pulls in Protobuf 2.5.0 via Flakka.

          commit b34a95c661aca2b301f3fef6df6adca2eb6a5e3c
          Author: Till Rohrmann <trohrmann@apache.org>
          Date: 2017-01-18T14:06:12Z

          FLINK-5508 [mesos] Introduce ZooKeeperUtilityFactory to create ZooKeeper utility classes

          This commit adds utility classes to abstract the CuratorFramework dependency from ZooKeeper
          utility classes away. That way it is possible for modules outside of flink-runtime to use
          these utility classes without facing the problem of a relocated curator dependency.


          Show
          githubbot ASF GitHub Bot added a comment - GitHub user tillrohrmann opened a pull request: https://github.com/apache/flink/pull/3158 [backport] FLINK-5508 FLINK-5496 FLINK-5495 Fix Mesos HA problems This PR is a backport of #3155, #3156 and #3157 onto the `release-1.2` branch. `da3358a` fixes: The ZooKeeperMesosWorkerStore instantiates a ZooKeeperStateHandleStore which requires an Executor instance. This executor is now given to the ZooKeeperMesosWorkerStore. `e14fb2d` fixes: Only relocate Mesos Protobuf dependency in flink-mesos. This avoids problems with Mesos because Flink pulls in Protobuf 2.5.0 via Flakka. `b34a95c` fixes: This commit adds utility classes to abstract the CuratorFramework dependency from ZooKeeper utility classes away. That way it is possible for modules outside of flink-runtime to use these utility classes without facing the problem of a relocated curator dependency. You can merge this pull request into a Git repository by running: $ git pull https://github.com/tillrohrmann/flink backportCuratorShading Alternatively you can review and apply these changes as the patch at: https://github.com/apache/flink/pull/3158.patch To close this pull request, make a commit to your master/trunk branch with (at least) the following in the commit message: This closes #3158 commit da3358ac5addac540910a9c492aece6435797b52 Author: Till Rohrmann <trohrmann@apache.org> Date: 2017-01-16T13:14:18Z FLINK-5495 [mesos] Provide executor to ZooKeeperMesosWorkerStore The ZooKeeperMesosWorkerStore instantiates a ZooKeeperStateHandleStore which requires an Executor instance. This executor is now given to the ZooKeeperMesosWorkerStore. commit e14fb2de0414378fb32aa21e88cd2ac0ba956da5 Author: Till Rohrmann <trohrmann@apache.org> Date: 2017-01-16T13:01:10Z FLINK-5496 [mesos] Relocate Mesos Protobuf dependency to avoid version conflicts Only relocate Mesos Protobuf dependency in flink-mesos. This avoids problems with Mesos because Flink pulls in Protobuf 2.5.0 via Flakka. commit b34a95c661aca2b301f3fef6df6adca2eb6a5e3c Author: Till Rohrmann <trohrmann@apache.org> Date: 2017-01-18T14:06:12Z FLINK-5508 [mesos] Introduce ZooKeeperUtilityFactory to create ZooKeeper utility classes This commit adds utility classes to abstract the CuratorFramework dependency from ZooKeeper utility classes away. That way it is possible for modules outside of flink-runtime to use these utility classes without facing the problem of a relocated curator dependency.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tzulitai commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97255571

          — Diff: flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosFlinkResourceManagerTest.java —
          @@ -73,13 +74,16 @@
          /**

          • General tests for the Mesos resource manager component.
            */
            -public class MesosFlinkResourceManagerTest {
            +public class MesosFlinkResourceManagerTest extends TestLogger {

          private static final Logger LOG = LoggerFactory.getLogger(MesosFlinkResourceManagerTest.class);

          private static ActorSystem system;

          • private static Configuration config = new Configuration() {{
            + private static Configuration config = new Configuration() {
            + private static final long serialVersionUID = -952579203067648838L;
            +
            + {
              • End diff –

          The indentation in this static block seems to be disordered (the following 2 `setInteger`s should be indented with one more tab).

          Show
          githubbot ASF GitHub Bot added a comment - Github user tzulitai commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97255571 — Diff: flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosFlinkResourceManagerTest.java — @@ -73,13 +74,16 @@ /** General tests for the Mesos resource manager component. */ -public class MesosFlinkResourceManagerTest { +public class MesosFlinkResourceManagerTest extends TestLogger { private static final Logger LOG = LoggerFactory.getLogger(MesosFlinkResourceManagerTest.class); private static ActorSystem system; private static Configuration config = new Configuration() {{ + private static Configuration config = new Configuration() { + private static final long serialVersionUID = -952579203067648838L; + + { End diff – The indentation in this static block seems to be disordered (the following 2 `setInteger`s should be indented with one more tab).
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tzulitai commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97256203

          — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperVersionedValue.java —
          @@ -0,0 +1,43 @@
          +/*
          + * 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.flink.runtime.zookeeper;
          +
          +import org.apache.curator.framework.recipes.shared.VersionedValue;
          +import org.apache.flink.util.Preconditions;
          +
          +/**
          + * Wrapper class for a

          {@link VersionedValue}

          so that we don't expose a curator dependency in our
          + * internal APIs. Such an exposure is problematic due to the relocation of curator.
          + */
          +public class ZooKeeperVersionedValue<T> {
          +
          + private final VersionedValue<T> versionedValue;
          +
          + public ZooKeeperVersionedValue(VersionedValue<T> versionedValue)

          { + this.versionedValue = Preconditions.checkNotNull(versionedValue); + }

          +
          + VersionedValue<T> getVersionedValue() {
          — End diff –

          nitpick: I would place the public methods before any other package-private / private access methods.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tzulitai commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97256203 — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperVersionedValue.java — @@ -0,0 +1,43 @@ +/* + * 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.flink.runtime.zookeeper; + +import org.apache.curator.framework.recipes.shared.VersionedValue; +import org.apache.flink.util.Preconditions; + +/** + * Wrapper class for a {@link VersionedValue} so that we don't expose a curator dependency in our + * internal APIs. Such an exposure is problematic due to the relocation of curator. + */ +public class ZooKeeperVersionedValue<T> { + + private final VersionedValue<T> versionedValue; + + public ZooKeeperVersionedValue(VersionedValue<T> versionedValue) { + this.versionedValue = Preconditions.checkNotNull(versionedValue); + } + + VersionedValue<T> getVersionedValue() { — End diff – nitpick: I would place the public methods before any other package-private / private access methods.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tzulitai commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97257309

          — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperUtilityFactory.java —
          @@ -0,0 +1,120 @@
          +/*
          + * 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.flink.runtime.zookeeper;
          +
          +import org.apache.curator.framework.CuratorFramework;
          +import org.apache.curator.framework.recipes.shared.SharedCount;
          +import org.apache.curator.framework.recipes.shared.SharedValue;
          +import org.apache.flink.configuration.Configuration;
          +import org.apache.flink.runtime.util.ZooKeeperUtils;
          +
          +import java.io.Serializable;
          +import java.util.concurrent.Executor;
          +
          +/**
          + * Creates ZooKeeper utility classes without exposing the

          {@link CuratorFramework}

          dependency. The
          + * curator framework is cached in this instance and shared among all created ZooKeeper utility
          + * instances. This requires that the utility classes DO NOT close the provided curator framework.
          + *
          + * <p>The curator framework is closed by calling the

          {@link #close(boolean)}

          method.
          + */
          +public class ZooKeeperUtilityFactory {
          +
          + private final CuratorFramework root;
          +
          + // Facade bound to the provided path
          + private final CuratorFramework facade;
          +
          + public ZooKeeperUtilityFactory(Configuration configuration, String path) throws Exception {
          — End diff –

          Should perform precondition checks for arguments.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tzulitai commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97257309 — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperUtilityFactory.java — @@ -0,0 +1,120 @@ +/* + * 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.flink.runtime.zookeeper; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.shared.SharedCount; +import org.apache.curator.framework.recipes.shared.SharedValue; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.util.ZooKeeperUtils; + +import java.io.Serializable; +import java.util.concurrent.Executor; + +/** + * Creates ZooKeeper utility classes without exposing the {@link CuratorFramework} dependency. The + * curator framework is cached in this instance and shared among all created ZooKeeper utility + * instances. This requires that the utility classes DO NOT close the provided curator framework. + * + * <p>The curator framework is closed by calling the {@link #close(boolean)} method. + */ +public class ZooKeeperUtilityFactory { + + private final CuratorFramework root; + + // Facade bound to the provided path + private final CuratorFramework facade; + + public ZooKeeperUtilityFactory(Configuration configuration, String path) throws Exception { — End diff – Should perform precondition checks for arguments.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tzulitai commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97257878

          — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperUtilityFactory.java —
          @@ -0,0 +1,120 @@
          +/*
          + * 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.flink.runtime.zookeeper;
          +
          +import org.apache.curator.framework.CuratorFramework;
          +import org.apache.curator.framework.recipes.shared.SharedCount;
          +import org.apache.curator.framework.recipes.shared.SharedValue;
          +import org.apache.flink.configuration.Configuration;
          +import org.apache.flink.runtime.util.ZooKeeperUtils;
          +
          +import java.io.Serializable;
          +import java.util.concurrent.Executor;
          +
          +/**
          + * Creates ZooKeeper utility classes without exposing the

          {@link CuratorFramework} dependency. The
          + * curator framework is cached in this instance and shared among all created ZooKeeper utility
          + * instances. This requires that the utility classes DO NOT close the provided curator framework.
          + *
          + * <p>The curator framework is closed by calling the {@link #close(boolean)} method.
          + */
          +public class ZooKeeperUtilityFactory {
          +
          + private final CuratorFramework root;
          +
          + // Facade bound to the provided path
          + private final CuratorFramework facade;
          +
          + public ZooKeeperUtilityFactory(Configuration configuration, String path) throws Exception { + root = ZooKeeperUtils.startCuratorFramework(configuration); + + root.newNamespaceAwareEnsurePath(path).ensure(root.getZookeeperClient()); + facade = root.usingNamespace(ZooKeeperUtils.generateZookeeperPath(root.getNamespace(), path)); + }
          +
          + /**
          + * Closes the ZooKeeperUtilityFactory. This entails closing the cached {@link CuratorFramework}

          + * instance. If cleanup is true, then the initial path and all its children are deleted.
          + *
          + * @param cleanup deletes the initial path and all of its children to clean up
          + * @throws Exception when deleting the znodes
          + */
          + public void close(boolean cleanup) throws Exception {
          + if (cleanup)

          { + facade.delete().deletingChildrenIfNeeded().forPath("/"); + }

          +
          + root.close();
          + }
          +
          + /**
          + * Creates a

          {@link ZooKeeperStateHandleStore}

          instance with the provided arguments.
          + *
          + * @param zkStateHandleStorePath specifying the path in ZooKeeper to store the state handles to
          + * @param stateStorageHelper storing the actual state data
          + * @param executor to run asynchronous callbacks of the state handle store
          + * @param <T> Type of the state to be stored
          + * @return a ZooKeeperStateHandleStore instance
          + * @throws Exception if ZooKeeper could not create the provided state handle store path in
          + * ZooKeeper
          — End diff –

          nit: Does this line in the Javadoc need to be split into 2 lines? I don't think it's exceeding the max character limit, is it?

          Show
          githubbot ASF GitHub Bot added a comment - Github user tzulitai commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97257878 — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperUtilityFactory.java — @@ -0,0 +1,120 @@ +/* + * 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.flink.runtime.zookeeper; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.shared.SharedCount; +import org.apache.curator.framework.recipes.shared.SharedValue; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.util.ZooKeeperUtils; + +import java.io.Serializable; +import java.util.concurrent.Executor; + +/** + * Creates ZooKeeper utility classes without exposing the {@link CuratorFramework} dependency. The + * curator framework is cached in this instance and shared among all created ZooKeeper utility + * instances. This requires that the utility classes DO NOT close the provided curator framework. + * + * <p>The curator framework is closed by calling the {@link #close(boolean)} method. + */ +public class ZooKeeperUtilityFactory { + + private final CuratorFramework root; + + // Facade bound to the provided path + private final CuratorFramework facade; + + public ZooKeeperUtilityFactory(Configuration configuration, String path) throws Exception { + root = ZooKeeperUtils.startCuratorFramework(configuration); + + root.newNamespaceAwareEnsurePath(path).ensure(root.getZookeeperClient()); + facade = root.usingNamespace(ZooKeeperUtils.generateZookeeperPath(root.getNamespace(), path)); + } + + /** + * Closes the ZooKeeperUtilityFactory. This entails closing the cached {@link CuratorFramework} + * instance. If cleanup is true, then the initial path and all its children are deleted. + * + * @param cleanup deletes the initial path and all of its children to clean up + * @throws Exception when deleting the znodes + */ + public void close(boolean cleanup) throws Exception { + if (cleanup) { + facade.delete().deletingChildrenIfNeeded().forPath("/"); + } + + root.close(); + } + + /** + * Creates a {@link ZooKeeperStateHandleStore} instance with the provided arguments. + * + * @param zkStateHandleStorePath specifying the path in ZooKeeper to store the state handles to + * @param stateStorageHelper storing the actual state data + * @param executor to run asynchronous callbacks of the state handle store + * @param <T> Type of the state to be stored + * @return a ZooKeeperStateHandleStore instance + * @throws Exception if ZooKeeper could not create the provided state handle store path in + * ZooKeeper — End diff – nit: Does this line in the Javadoc need to be split into 2 lines? I don't think it's exceeding the max character limit, is it?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tzulitai commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97255788

          — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java —
          @@ -376,15 +376,19 @@ public static ZooKeeperCheckpointIDCounter createCheckpointIDCounter(
          }
          }

          • private static String generateZookeeperPath(String root, String namespace) {
            + public static String generateZookeeperPath(String root, String namespace) {
            if (!namespace.startsWith("/")) { - namespace = "/" + namespace; + namespace = '/' + namespace; }

          if (namespace.endsWith("/"))

          { namespace = namespace.substring(0, namespace.length() - 1); }

          + if (root.endsWith("/")) {
          — End diff –

          Nice catch ..

          Show
          githubbot ASF GitHub Bot added a comment - Github user tzulitai commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97255788 — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/util/ZooKeeperUtils.java — @@ -376,15 +376,19 @@ public static ZooKeeperCheckpointIDCounter createCheckpointIDCounter( } } private static String generateZookeeperPath(String root, String namespace) { + public static String generateZookeeperPath(String root, String namespace) { if (!namespace.startsWith("/")) { - namespace = "/" + namespace; + namespace = '/' + namespace; } if (namespace.endsWith("/")) { namespace = namespace.substring(0, namespace.length() - 1); } + if (root.endsWith("/")) { — End diff – Nice catch ..
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tzulitai commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97253388

          — Diff: flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java —
          @@ -394,6 +395,14 @@ protected int runPrivileged(Configuration config, Configuration dynamicPropertie
          }
          }

          + if (mesosServices != null) {
          + try

          { + mesosServices.close(false); + }

          catch (Throwable tt) {
          + LOG.error("Error closing the ZooKeeperUtilityFactory.", tt);
          — End diff –

          This error message is only relevant to `ZooKeeperMesosServices`, and not `StandaloneMesosServices.`
          Should we just use "Failed to clean up and close MesosServices." here, and rely on the stack trace in logs to find out the case?

          Show
          githubbot ASF GitHub Bot added a comment - Github user tzulitai commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97253388 — Diff: flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java — @@ -394,6 +395,14 @@ protected int runPrivileged(Configuration config, Configuration dynamicPropertie } } + if (mesosServices != null) { + try { + mesosServices.close(false); + } catch (Throwable tt) { + LOG.error("Error closing the ZooKeeperUtilityFactory.", tt); — End diff – This error message is only relevant to `ZooKeeperMesosServices`, and not `StandaloneMesosServices.` Should we just use "Failed to clean up and close MesosServices." here, and rely on the stack trace in logs to find out the case?
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tzulitai commented on the issue:

          https://github.com/apache/flink/pull/3158

          This is an identical backport / forwardport of the other 3 Mesos fixes, correct?
          If so, since #3155 and #3156 all have +1s, and #3157 is also a +1 once the minor code style comments are addressed, this is also a +1 from my side.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tzulitai commented on the issue: https://github.com/apache/flink/pull/3158 This is an identical backport / forwardport of the other 3 Mesos fixes, correct? If so, since #3155 and #3156 all have +1s, and #3157 is also a +1 once the minor code style comments are addressed, this is also a +1 from my side.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tillrohrmann commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97273510

          — Diff: flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java —
          @@ -394,6 +395,14 @@ protected int runPrivileged(Configuration config, Configuration dynamicPropertie
          }
          }

          + if (mesosServices != null) {
          + try

          { + mesosServices.close(false); + }

          catch (Throwable tt) {
          + LOG.error("Error closing the ZooKeeperUtilityFactory.", tt);
          — End diff –

          True, will change it.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tillrohrmann commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97273510 — Diff: flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/MesosApplicationMasterRunner.java — @@ -394,6 +395,14 @@ protected int runPrivileged(Configuration config, Configuration dynamicPropertie } } + if (mesosServices != null) { + try { + mesosServices.close(false); + } catch (Throwable tt) { + LOG.error("Error closing the ZooKeeperUtilityFactory.", tt); — End diff – True, will change it.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tillrohrmann commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97273631

          — Diff: flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosFlinkResourceManagerTest.java —
          @@ -73,13 +74,16 @@
          /**

          • General tests for the Mesos resource manager component.
            */
            -public class MesosFlinkResourceManagerTest {
            +public class MesosFlinkResourceManagerTest extends TestLogger {

          private static final Logger LOG = LoggerFactory.getLogger(MesosFlinkResourceManagerTest.class);

          private static ActorSystem system;

          • private static Configuration config = new Configuration() {{
            + private static Configuration config = new Configuration() {
            + private static final long serialVersionUID = -952579203067648838L;
            +
            + {
              • End diff –

          Good catch. Will change it.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tillrohrmann commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97273631 — Diff: flink-mesos/src/test/java/org/apache/flink/mesos/runtime/clusterframework/MesosFlinkResourceManagerTest.java — @@ -73,13 +74,16 @@ /** General tests for the Mesos resource manager component. */ -public class MesosFlinkResourceManagerTest { +public class MesosFlinkResourceManagerTest extends TestLogger { private static final Logger LOG = LoggerFactory.getLogger(MesosFlinkResourceManagerTest.class); private static ActorSystem system; private static Configuration config = new Configuration() {{ + private static Configuration config = new Configuration() { + private static final long serialVersionUID = -952579203067648838L; + + { End diff – Good catch. Will change it.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tillrohrmann commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97273730

          — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperVersionedValue.java —
          @@ -0,0 +1,43 @@
          +/*
          + * 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.flink.runtime.zookeeper;
          +
          +import org.apache.curator.framework.recipes.shared.VersionedValue;
          +import org.apache.flink.util.Preconditions;
          +
          +/**
          + * Wrapper class for a

          {@link VersionedValue}

          so that we don't expose a curator dependency in our
          + * internal APIs. Such an exposure is problematic due to the relocation of curator.
          + */
          +public class ZooKeeperVersionedValue<T> {
          +
          + private final VersionedValue<T> versionedValue;
          +
          + public ZooKeeperVersionedValue(VersionedValue<T> versionedValue)

          { + this.versionedValue = Preconditions.checkNotNull(versionedValue); + }

          +
          + VersionedValue<T> getVersionedValue() {
          — End diff –

          Good point. Will change it.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tillrohrmann commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97273730 — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperVersionedValue.java — @@ -0,0 +1,43 @@ +/* + * 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.flink.runtime.zookeeper; + +import org.apache.curator.framework.recipes.shared.VersionedValue; +import org.apache.flink.util.Preconditions; + +/** + * Wrapper class for a {@link VersionedValue} so that we don't expose a curator dependency in our + * internal APIs. Such an exposure is problematic due to the relocation of curator. + */ +public class ZooKeeperVersionedValue<T> { + + private final VersionedValue<T> versionedValue; + + public ZooKeeperVersionedValue(VersionedValue<T> versionedValue) { + this.versionedValue = Preconditions.checkNotNull(versionedValue); + } + + VersionedValue<T> getVersionedValue() { — End diff – Good point. Will change it.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tillrohrmann commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97273823

          — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperUtilityFactory.java —
          @@ -0,0 +1,120 @@
          +/*
          + * 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.flink.runtime.zookeeper;
          +
          +import org.apache.curator.framework.CuratorFramework;
          +import org.apache.curator.framework.recipes.shared.SharedCount;
          +import org.apache.curator.framework.recipes.shared.SharedValue;
          +import org.apache.flink.configuration.Configuration;
          +import org.apache.flink.runtime.util.ZooKeeperUtils;
          +
          +import java.io.Serializable;
          +import java.util.concurrent.Executor;
          +
          +/**
          + * Creates ZooKeeper utility classes without exposing the

          {@link CuratorFramework}

          dependency. The
          + * curator framework is cached in this instance and shared among all created ZooKeeper utility
          + * instances. This requires that the utility classes DO NOT close the provided curator framework.
          + *
          + * <p>The curator framework is closed by calling the

          {@link #close(boolean)}

          method.
          + */
          +public class ZooKeeperUtilityFactory {
          +
          + private final CuratorFramework root;
          +
          + // Facade bound to the provided path
          + private final CuratorFramework facade;
          +
          + public ZooKeeperUtilityFactory(Configuration configuration, String path) throws Exception {
          — End diff –

          Yes indeed. Will change it.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tillrohrmann commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97273823 — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperUtilityFactory.java — @@ -0,0 +1,120 @@ +/* + * 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.flink.runtime.zookeeper; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.shared.SharedCount; +import org.apache.curator.framework.recipes.shared.SharedValue; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.util.ZooKeeperUtils; + +import java.io.Serializable; +import java.util.concurrent.Executor; + +/** + * Creates ZooKeeper utility classes without exposing the {@link CuratorFramework} dependency. The + * curator framework is cached in this instance and shared among all created ZooKeeper utility + * instances. This requires that the utility classes DO NOT close the provided curator framework. + * + * <p>The curator framework is closed by calling the {@link #close(boolean)} method. + */ +public class ZooKeeperUtilityFactory { + + private final CuratorFramework root; + + // Facade bound to the provided path + private final CuratorFramework facade; + + public ZooKeeperUtilityFactory(Configuration configuration, String path) throws Exception { — End diff – Yes indeed. Will change it.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tillrohrmann commented on a diff in the pull request:

          https://github.com/apache/flink/pull/3157#discussion_r97274290

          — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperUtilityFactory.java —
          @@ -0,0 +1,120 @@
          +/*
          + * 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.flink.runtime.zookeeper;
          +
          +import org.apache.curator.framework.CuratorFramework;
          +import org.apache.curator.framework.recipes.shared.SharedCount;
          +import org.apache.curator.framework.recipes.shared.SharedValue;
          +import org.apache.flink.configuration.Configuration;
          +import org.apache.flink.runtime.util.ZooKeeperUtils;
          +
          +import java.io.Serializable;
          +import java.util.concurrent.Executor;
          +
          +/**
          + * Creates ZooKeeper utility classes without exposing the

          {@link CuratorFramework} dependency. The
          + * curator framework is cached in this instance and shared among all created ZooKeeper utility
          + * instances. This requires that the utility classes DO NOT close the provided curator framework.
          + *
          + * <p>The curator framework is closed by calling the {@link #close(boolean)} method.
          + */
          +public class ZooKeeperUtilityFactory {
          +
          + private final CuratorFramework root;
          +
          + // Facade bound to the provided path
          + private final CuratorFramework facade;
          +
          + public ZooKeeperUtilityFactory(Configuration configuration, String path) throws Exception { + root = ZooKeeperUtils.startCuratorFramework(configuration); + + root.newNamespaceAwareEnsurePath(path).ensure(root.getZookeeperClient()); + facade = root.usingNamespace(ZooKeeperUtils.generateZookeeperPath(root.getNamespace(), path)); + }
          +
          + /**
          + * Closes the ZooKeeperUtilityFactory. This entails closing the cached {@link CuratorFramework}

          + * instance. If cleanup is true, then the initial path and all its children are deleted.
          + *
          + * @param cleanup deletes the initial path and all of its children to clean up
          + * @throws Exception when deleting the znodes
          + */
          + public void close(boolean cleanup) throws Exception {
          + if (cleanup)

          { + facade.delete().deletingChildrenIfNeeded().forPath("/"); + }

          +
          + root.close();
          + }
          +
          + /**
          + * Creates a

          {@link ZooKeeperStateHandleStore}

          instance with the provided arguments.
          + *
          + * @param zkStateHandleStorePath specifying the path in ZooKeeper to store the state handles to
          + * @param stateStorageHelper storing the actual state data
          + * @param executor to run asynchronous callbacks of the state handle store
          + * @param <T> Type of the state to be stored
          + * @return a ZooKeeperStateHandleStore instance
          + * @throws Exception if ZooKeeper could not create the provided state handle store path in
          + * ZooKeeper
          — End diff –

          As far as I know, we don't have a hard line length limit specified for Java. Personally, I try to keep it around 100 chars per line because this is in line with the Scala style. But it really depends on the statement and totally subjective here...

          Show
          githubbot ASF GitHub Bot added a comment - Github user tillrohrmann commented on a diff in the pull request: https://github.com/apache/flink/pull/3157#discussion_r97274290 — Diff: flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperUtilityFactory.java — @@ -0,0 +1,120 @@ +/* + * 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.flink.runtime.zookeeper; + +import org.apache.curator.framework.CuratorFramework; +import org.apache.curator.framework.recipes.shared.SharedCount; +import org.apache.curator.framework.recipes.shared.SharedValue; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.runtime.util.ZooKeeperUtils; + +import java.io.Serializable; +import java.util.concurrent.Executor; + +/** + * Creates ZooKeeper utility classes without exposing the {@link CuratorFramework} dependency. The + * curator framework is cached in this instance and shared among all created ZooKeeper utility + * instances. This requires that the utility classes DO NOT close the provided curator framework. + * + * <p>The curator framework is closed by calling the {@link #close(boolean)} method. + */ +public class ZooKeeperUtilityFactory { + + private final CuratorFramework root; + + // Facade bound to the provided path + private final CuratorFramework facade; + + public ZooKeeperUtilityFactory(Configuration configuration, String path) throws Exception { + root = ZooKeeperUtils.startCuratorFramework(configuration); + + root.newNamespaceAwareEnsurePath(path).ensure(root.getZookeeperClient()); + facade = root.usingNamespace(ZooKeeperUtils.generateZookeeperPath(root.getNamespace(), path)); + } + + /** + * Closes the ZooKeeperUtilityFactory. This entails closing the cached {@link CuratorFramework} + * instance. If cleanup is true, then the initial path and all its children are deleted. + * + * @param cleanup deletes the initial path and all of its children to clean up + * @throws Exception when deleting the znodes + */ + public void close(boolean cleanup) throws Exception { + if (cleanup) { + facade.delete().deletingChildrenIfNeeded().forPath("/"); + } + + root.close(); + } + + /** + * Creates a {@link ZooKeeperStateHandleStore} instance with the provided arguments. + * + * @param zkStateHandleStorePath specifying the path in ZooKeeper to store the state handles to + * @param stateStorageHelper storing the actual state data + * @param executor to run asynchronous callbacks of the state handle store + * @param <T> Type of the state to be stored + * @return a ZooKeeperStateHandleStore instance + * @throws Exception if ZooKeeper could not create the provided state handle store path in + * ZooKeeper — End diff – As far as I know, we don't have a hard line length limit specified for Java. Personally, I try to keep it around 100 chars per line because this is in line with the Scala style. But it really depends on the statement and totally subjective here...
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tillrohrmann commented on the issue:

          https://github.com/apache/flink/pull/3157

          Thanks for the review @tzulitai. I'll address your comments and then merge this PR.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tillrohrmann commented on the issue: https://github.com/apache/flink/pull/3157 Thanks for the review @tzulitai. I'll address your comments and then merge this PR.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tillrohrmann commented on the issue:

          https://github.com/apache/flink/pull/3158

          Thanks for the review @tzulitai. I will address your comments and then merge this PR.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tillrohrmann commented on the issue: https://github.com/apache/flink/pull/3158 Thanks for the review @tzulitai. I will address your comments and then merge this PR.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user asfgit closed the pull request at:

          https://github.com/apache/flink/pull/3157

          Show
          githubbot ASF GitHub Bot added a comment - Github user asfgit closed the pull request at: https://github.com/apache/flink/pull/3157
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tillrohrmann commented on the issue:

          https://github.com/apache/flink/pull/3158

          Merged to the `release-1.2` branch.

          Show
          githubbot ASF GitHub Bot added a comment - Github user tillrohrmann commented on the issue: https://github.com/apache/flink/pull/3158 Merged to the `release-1.2` branch.
          Hide
          githubbot ASF GitHub Bot added a comment -

          Github user tillrohrmann closed the pull request at:

          https://github.com/apache/flink/pull/3158

          Show
          githubbot ASF GitHub Bot added a comment - Github user tillrohrmann closed the pull request at: https://github.com/apache/flink/pull/3158
          Hide
          till.rohrmann Till Rohrmann added a comment -

          1.3.0: Fixed via df3f11979ed2895ed548766bac061e6cda8f6881
          1.2.0: Fixed via 9fc1fe01798537b623a1a3797e2e8c0967d4673c

          Show
          till.rohrmann Till Rohrmann added a comment - 1.3.0: Fixed via df3f11979ed2895ed548766bac061e6cda8f6881 1.2.0: Fixed via 9fc1fe01798537b623a1a3797e2e8c0967d4673c

            People

            • Assignee:
              till.rohrmann Till Rohrmann
              Reporter:
              till.rohrmann Till Rohrmann
            • Votes:
              0 Vote for this issue
              Watchers:
              2 Start watching this issue

              Dates

              • Created:
                Updated:
                Resolved:

                Development