Replace the use of Paths.get(...).getParent() for a zookeeper path in…

… ZooKeeperCache (apache#7558)

because it's system dependent and won't work when running Pulsar in Windows

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZkUtils.java

@@ -75,4 +75,46 @@ public static boolean checkNodeAndWaitExpired(ZooKeeper zk,
         }
     }
 
+    /**
+     * Returns the parent path of the provided path.
+     * Works for UNIX-style paths only and is intended to be used for zookeeper paths.
+     * If you have a file system path use {@link java.nio.file.Paths#get(String, String...)}
+     * and {@link java.nio.file.Path#getParent()} instead.
+     *
+     * @param path the zookeeper path
+     * @return the parent path or null if no parent exists
+     */
+    public static String getParentForPath(final String path) {
+        if (path == null) return null;
+        final int length = path.length();
+        if (length == 0 || (length == 1 && path.charAt(0) == '/')) return null;
+
+        int partStartIndex = 0;
+        char lastChar = path.charAt(0);
+        String lastPart = "/";  // needed, if it's an absolute path
+        final StringBuilder sb = new StringBuilder();
+        for (int index = 1; index < length; index++) {
+            final char c = path.charAt(index);
+            if (lastChar != '/') {
+                if (c == '/') {
+                    // First '/' after a non-'/' sequence defines the end of a part;
+                    // save the part for later addition (when it's clear it is part
+                    // of the parent path)
+                    lastPart = path.substring(partStartIndex, index);
+                    // Only needed, if it's the first part of a relative path to ensure
+                    // that the next part includes the leading '/'
+                    partStartIndex = -1;
+                }
+            } else if (c != '/') {
+                // First non-'/' after a (series of) '/' indicates that there is a new part
+                // after the saved, so the saved part definitely belongs to the parent path
+                sb.append(lastPart);
+                // Include the preceding '/' except for the first part
+                partStartIndex = (partStartIndex == 0) ? index : index - 1;
+            }
+            lastChar = c;
+        }
+        final String parentPath = sb.toString();
+        return (parentPath.length() == 0) ? null : parentPath;
+    }
 }

pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/ZooKeeperCache.java

@@ -25,7 +25,6 @@
 import com.google.common.collect.Sets;
 
 import java.io.IOException;
-import java.nio.file.Paths;
 import java.util.AbstractMap.SimpleImmutableEntry;
 import java.util.Collections;
 import java.util.Map;
@@ -146,7 +145,7 @@ public <T> void process(WatchedEvent event, final CacheUpdater<T> updater) {
             // ZookeeperCache instance then ZkChildrenCache may not invalidate for it's parent. Therefore, invalidate
             // cache for parent if child is created/deleted
             if (event.getType().equals(EventType.NodeCreated) || event.getType().equals(EventType.NodeDeleted)) {
-                childrenCache.synchronous().invalidate(Paths.get(path).getParent().toString());
+                childrenCache.synchronous().invalidate(ZkUtils.getParentForPath(path));
             }
             existsCache.synchronous().invalidate(path);
             if (executor != null && updater != null) {

pulsar-zookeeper-utils/src/test/java/org/apache/pulsar/zookeeper/ZkUtilsParentPathTest.java

@@ -0,0 +1,112 @@
+/**
+ * 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.pulsar.zookeeper;
+
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+import static org.testng.Assert.assertEquals;
+
+public class ZkUtilsParentPathTest {
+    @DataProvider(name = "pathsAndParents")
+    public static Object[][] pathsAndParents() {
+        return new Object[][] {
+                { null, null },
+                { "", null },
+                { "/", null },
+                { "//", null },
+                { "///", null },
+
+                { "a", null },
+                { "a/", null },
+                { "a//", null },
+                { "a///", null },
+
+                { "/a", "/" },
+                { "//a", "/" },
+                { "///a", "/" },
+                { "/a/", "/" },
+                { "/a//", "/" },
+                { "/a///", "/" },
+                { "//a/", "/" },
+                { "//a//", "/" },
+                { "//a///", "/" },
+                { "///a/", "/" },
+                { "///a//", "/" },
+                { "///a///", "/" },
+
+                { "/a/b", "/a" },
+                { "//a/b", "/a" },
+                { "///a/b", "/a" },
+                { "/a//b", "/a" },
+                { "/a///b", "/a" },
+                { "/a/b/", "/a" },
+                { "/a/b//", "/a" },
+                { "/a/b///", "/a" },
+                { "/a//b/", "/a" },
+                { "/a///b/", "/a" },
+
+                { "a/b", "a" },
+                { "a//b", "a" },
+                { "a///b", "a" },
+                { "a/b/", "a" },
+                { "a/b//", "a" },
+                { "a/b///", "a" },
+                { "a//b/", "a" },
+                { "a///b/", "a" },
+
+                { "/a/b/c", "/a/b" },
+                { "//a/b/c", "/a/b" },
+                { "///a/b/c", "/a/b" },
+                { "/a//b/c", "/a/b" },
+                { "/a///b/c", "/a/b" },
+                { "/a/b//c", "/a/b" },
+                { "/a/b///c", "/a/b" },
+                { "/a//b//c/", "/a/b" },
+                { "/a//b///c/", "/a/b" },
+                { "/a///b//c/", "/a/b" },
+                { "/a///b///c/", "/a/b" },
+                { "/a/b/c/", "/a/b" },
+                { "/a/b//c/", "/a/b" },
+                { "/a/b///c/", "/a/b" },
+
+                { "a/b/c", "a/b" },
+                { "a//b/c", "a/b" },
+                { "a///b/c", "a/b" },
+
+                { "abc", null },
+                { "abc/", null },
+                { "/abc", "/" },
+                { "/abc/", "/" },
+                { "/abc/def", "/abc" },
+                { "/abc/def/", "/abc" },
+                { "abc/def", "abc" },
+                { "abc/def/", "abc" },
+                { "/abc/def/ghi", "/abc/def" },
+                { "/abc/def/ghi/", "/abc/def" },
+                { "abc/def/ghi", "abc/def" },
+                { "abc/def/ghi/", "abc/def" }
+        };
+    }
+
+    @Test(dataProvider = "pathsAndParents")
+    public void testGetParentForPath(final String path, final String expectedParent) {
+        assertEquals(ZkUtils.getParentForPath(path), expectedParent, "Parent for " + path);
+    }
+}