[FLINK-7778] [build] Shade ZooKeeper dependency (followups)

  - Rename the 'flink-shaded-curator-recipes' module to 'flink-shaded-curator',
    because it actually contains more curator code than just the recipes.

  - Move the exception handling logic of 'ZooKeeperAccess' directly into the
    ZooKeeperStateHandleStore

flink-mesos/src/main/java/org/apache/flink/mesos/runtime/clusterframework/store/ZooKeeperMesosWorkerStore.java

@@ -25,10 +25,10 @@
 import org.apache.flink.runtime.zookeeper.ZooKeeperSharedValue;
 import org.apache.flink.runtime.zookeeper.ZooKeeperStateHandleStore;
 import org.apache.flink.runtime.zookeeper.ZooKeeperVersionedValue;
-import org.apache.flink.runtime.zookeeper.ZookeeperAccess;
 import org.apache.flink.util.FlinkException;
 
 import org.apache.mesos.Protos;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -214,19 +214,11 @@ public void putWorker(MesosWorkerStore.Worker worker) throws Exception {
 
 			int currentVersion = workersInZooKeeper.exists(path);
 			if (currentVersion == -1) {
-				try {
-					workersInZooKeeper.addAndLock(path, worker);
-					LOG.debug("Added {} in ZooKeeper.", worker);
-				} catch (Exception ex) {
-					throw ZookeeperAccess.wrapIfZooKeeperNodeExistsException(ex);
-				}
+				workersInZooKeeper.addAndLock(path, worker);
+				LOG.debug("Added {} in ZooKeeper.", worker);
 			} else {
-				try {
-					workersInZooKeeper.replace(path, currentVersion, worker);
-					LOG.debug("Updated {} in ZooKeeper.", worker);
-				} catch (Exception ex) {
-					throw ZookeeperAccess.wrapIfZooKeeperNoNodeException(ex);
-				}
+				workersInZooKeeper.replace(path, currentVersion, worker);
+				LOG.debug("Updated {} in ZooKeeper.", worker);
 			}
 		}
 	}

flink-runtime/src/main/java/org/apache/flink/runtime/zookeeper/ZooKeeperStateHandleStore.java

@@ -41,6 +41,7 @@
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.ConcurrentModificationException;
 import java.util.List;
 import java.util.UUID;
 import java.util.concurrent.Executor;
@@ -163,6 +164,9 @@ public RetrievableStateHandle<T> addAndLock(
 			success = true;
 			return storeHandle;
 		}
+		catch (KeeperException.NodeExistsException e) {
+			throw new ConcurrentModificationException("ZooKeeper unexpectedly modified", e);
+		}
 		finally {
 			if (!success) {
 				// Cleanup the state handle if it was not written to ZooKeeper.
@@ -202,8 +206,10 @@ public void replace(String pathInZooKeeper, int expectedVersion, T state) throws
 					.withVersion(expectedVersion)
 					.forPath(path, serializedStateHandle);
 			success = true;
+		} catch (KeeperException.NoNodeException e) {
+			throw new ConcurrentModificationException("ZooKeeper unexpectedly modified", e);
 		} finally {
-			if(success) {
+			if (success) {
 				oldStateHandle.discardState();
 			} else {
 				newStateHandle.discardState();
@@ -673,7 +679,7 @@ public void processResult(CuratorFramework client, CuratorEvent event) throws Ex
 			}
 
 		}
-	};
+	}
 
 	/**
 	 * Callback interface for remove calls

flink-shaded-curator-recipes/pom.xml → flink-shaded-curator/pom.xml

@@ -29,8 +29,8 @@ under the License.
 		<relativePath>..</relativePath>
 	</parent>
 
-	<artifactId>flink-shaded-curator-recipes</artifactId>
-	<name>flink-shaded-curator-recipes</name>
+	<artifactId>flink-shaded-curator</artifactId>
+	<name>flink-shaded-curator</name>
 
 	<packaging>jar</packaging>
 

pom.xml

@@ -54,7 +54,7 @@ under the License.
 		<module>tools/force-shading</module>
 		<module>flink-annotations</module>
 		<module>flink-shaded-hadoop</module>
-		<module>flink-shaded-curator-recipes</module>
+		<module>flink-shaded-curator</module>
 		<module>flink-core</module>
 		<module>flink-java</module>
 		<module>flink-java8</module>