Package management bookkeeper storage implementation (apache#8744)

Master Issue: apache#8676

### Motivation

Support bookkeeper storage for packages management service.

### Modifications

- Add bookkeeper storage provider
- Add bookkeeper storage unit tests

pulsar-package-management/bookkeeper-storage/pom.xml

@@ -0,0 +1,81 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+    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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <parent>
+        <artifactId>pulsar-package-management</artifactId>
+        <groupId>org.apache.pulsar</groupId>
+        <version>2.8.0-SNAPSHOT</version>
+    </parent>
+    <modelVersion>4.0.0</modelVersion>
+
+    <artifactId>bookkeeper-storage</artifactId>
+    <name>Apache Pulsar :: Package Management :: BookKeeper Storage</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>pulsar-package-core</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.distributedlog</groupId>
+            <artifactId>distributedlog-core</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>net.jpountz.lz4</groupId>
+                    <artifactId>lz4</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.zookeeper</groupId>
+            <artifactId>zookeeper</artifactId>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>managed-ledger</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>testmocks</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+
+</project>

pulsar-package-management/bookkeeper-storage/src/main/java/org/apache/pulsar/packages/management/storage/bookkeeper/BookKeeperPackagesStorage.java

@@ -0,0 +1,176 @@
+/**
+ * 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.packages.management.storage.bookkeeper;
+
+import com.google.common.base.Strings;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.distributedlog.DistributedLogConfiguration;
+import org.apache.distributedlog.DistributedLogConstants;
+import org.apache.distributedlog.api.DistributedLogManager;
+import org.apache.distributedlog.api.namespace.Namespace;
+import org.apache.distributedlog.api.namespace.NamespaceBuilder;
+import org.apache.distributedlog.exceptions.ZKException;
+import org.apache.distributedlog.impl.metadata.BKDLConfig;
+import org.apache.distributedlog.metadata.DLMetadata;
+import org.apache.distributedlog.namespace.NamespaceDriver;
+import org.apache.pulsar.packages.management.core.PackagesStorage;
+import org.apache.pulsar.packages.management.core.PackagesStorageConfiguration;
+import org.apache.zookeeper.KeeperException;
+
+
+/**
+ * Packages management storage implementation with bookkeeper.
+ */
+@Slf4j
+public class BookKeeperPackagesStorage implements PackagesStorage {
+
+    private final static String NS_CLIENT_ID = "packages-management";
+    final BookKeeperPackagesStorageConfiguration configuration;
+    private Namespace namespace;
+
+    BookKeeperPackagesStorage(PackagesStorageConfiguration configuration) {
+        this.configuration = new BookKeeperPackagesStorageConfiguration(configuration);
+    }
+
+    @Override
+    public void initialize() {
+        DistributedLogConfiguration conf = new DistributedLogConfiguration()
+            .setImmediateFlushEnabled(true)
+            .setOutputBufferSize(0)
+            .setWriteQuorumSize(configuration.getNumReplicas())
+            .setEnsembleSize(configuration.getNumReplicas())
+            .setAckQuorumSize(configuration.getNumReplicas())
+            .setLockTimeout(DistributedLogConstants.LOCK_IMMEDIATE);
+        if (!Strings.isNullOrEmpty(configuration.getBookkeeperClientAuthenticationPlugin())) {
+            conf.setProperty("bkc.clientAuthProviderFactoryClass",
+                configuration.getBookkeeperClientAuthenticationPlugin());
+            if (!Strings.isNullOrEmpty(configuration.getBookkeeperClientAuthenticationParametersName())) {
+                conf.setProperty("bkc." + configuration.getBookkeeperClientAuthenticationParametersName(),
+                    configuration.getBookkeeperClientAuthenticationParameters());
+            }
+        }
+        try {
+            this.namespace = NamespaceBuilder.newBuilder()
+                .conf(conf).clientId(NS_CLIENT_ID).uri(initializeDlogNamespace()).build();
+        } catch (IOException e) {
+            throw new RuntimeException("Initialize distributed log for packages management service failed.", e);
+        }
+        log.info("Packages management bookKeeper storage initialized successfully");
+    }
+
+    private URI initializeDlogNamespace() throws IOException {
+        BKDLConfig bkdlConfig = new BKDLConfig(configuration.getZkServers(), configuration.getLedgersRootPath());
+        DLMetadata dlMetadata = DLMetadata.create(bkdlConfig);
+        URI dlogURI = URI.create(String.format("distributedlog://%s/pulsar/packages", configuration.getZkServers()));
+        try {
+            dlMetadata.create(dlogURI);
+        } catch (ZKException e) {
+            if (e.getKeeperExceptionCode() == KeeperException.Code.NODEEXISTS) {
+                return dlogURI;
+            }
+        }
+        return dlogURI;
+    }
+
+    private CompletableFuture<DistributedLogManager> openLogManagerAsync(String path) {
+        CompletableFuture<DistributedLogManager> logFuture = new CompletableFuture<>();
+        CompletableFuture.runAsync(() -> {
+            try {
+                logFuture.complete(namespace.openLog(path));
+            } catch (IOException e) {
+                logFuture.completeExceptionally(e);
+            }
+        });
+        return logFuture;
+    }
+
+    @Override
+    public CompletableFuture<Void> writeAsync(String path, InputStream inputStream) {
+        return openLogManagerAsync(path)
+            .thenCompose(DLOutputStream::openWriterAsync)
+            .thenCompose(dlOutputStream -> dlOutputStream.writeAsync(inputStream))
+            .thenCompose(DLOutputStream::closeAsync);
+    }
+
+    @Override
+    public CompletableFuture<Void> readAsync(String path, OutputStream outputStream) {
+        return openLogManagerAsync(path)
+            .thenCompose(DLInputStream::openReaderAsync)
+            .thenCompose(dlInputStream -> dlInputStream.readAsync(outputStream))
+            .thenCompose(DLInputStream::closeAsync);
+    }
+
+    @Override
+    public CompletableFuture<Void> deleteAsync(String path) {
+        return namespace.getNamespaceDriver().getLogMetadataStore().getLogLocation(path)
+            .thenCompose(uri -> uri.map(value -> namespace.getNamespaceDriver()
+                .getLogStreamMetadataStore(NamespaceDriver.Role.WRITER).deleteLog(value, path))
+                .orElse(null));
+    }
+
+
+    @Override
+    public CompletableFuture<List<String>> listAsync(String path) {
+        return namespace.getNamespaceDriver().getLogMetadataStore().getLogs(path)
+            .thenApply(logs -> {
+                ArrayList<String> packages = new ArrayList<>();
+                logs.forEachRemaining(packages::add);
+                return packages;
+            });
+    }
+
+    @Override
+    public CompletableFuture<Boolean> existAsync(String path) {
+        CompletableFuture<Boolean> result = new CompletableFuture<>();
+        namespace.getNamespaceDriver().getLogMetadataStore().getLogLocation(path)
+            .whenComplete((uriOptional, throwable) -> {
+                if (throwable != null) {
+                    result.complete(false);
+                    return;
+                }
+
+                if (uriOptional.isPresent()) {
+                    namespace.getNamespaceDriver()
+                        .getLogStreamMetadataStore(NamespaceDriver.Role.WRITER)
+                        .logExists(uriOptional.get(), path)
+                        .whenComplete((ignore, e) -> {
+                            if (e != null) {
+                                result.complete(false);
+                            } else {
+                                result.complete(true);
+                            }
+                        });
+                } else {
+                    result.complete(false);
+                }
+            });
+        return result;    }
+
+    @Override
+    public CompletableFuture<Void> closeAsync() {
+        return CompletableFuture.runAsync(() -> this.namespace.close());
+    }
+}

pulsar-package-management/bookkeeper-storage/src/main/java/org/apache/pulsar/packages/management/storage/bookkeeper/BookKeeperPackagesStorageConfiguration.java

@@ -0,0 +1,70 @@
+/**
+ * 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.packages.management.storage.bookkeeper;
+
+import org.apache.pulsar.packages.management.core.PackagesStorageConfiguration;
+import org.apache.pulsar.packages.management.core.impl.DefaultPackagesStorageConfiguration;
+
+public class BookKeeperPackagesStorageConfiguration implements PackagesStorageConfiguration {
+
+    private final PackagesStorageConfiguration configuration;
+
+    BookKeeperPackagesStorageConfiguration() {
+        this.configuration = new DefaultPackagesStorageConfiguration();
+    }
+
+    BookKeeperPackagesStorageConfiguration(PackagesStorageConfiguration configuration) {
+        this.configuration = configuration;
+    }
+
+    int getNumReplicas() {
+        return Integer.parseInt(getProperty("numReplicas"));
+    }
+
+    String getZkServers() {
+        return getProperty("zkServers");
+    }
+
+    String getLedgersRootPath() {
+        return getProperty("ledgerRootPath");
+    }
+
+    String getBookkeeperClientAuthenticationPlugin() {
+        return getProperty("bookkeeperClientAuthenticationPlugin");
+    }
+
+    String getBookkeeperClientAuthenticationParametersName() {
+        return getProperty("bookkeeperClientAuthenticationParametersName");
+    }
+
+    String getBookkeeperClientAuthenticationParameters() {
+        return getProperty("bookkeeperClientAuthenticationParameters");
+    }
+
+
+    @Override
+    public String getProperty(String key) {
+        return configuration.getProperty(key);
+    }
+
+    @Override
+    public void setProperty(String key, String value) {
+        configuration.setProperty(key, value);
+    }
+}

pulsar-package-management/bookkeeper-storage/src/main/java/org/apache/pulsar/packages/management/storage/bookkeeper/BookKeeperPackagesStorageProvider.java

@@ -0,0 +1,30 @@
+/**
+ * 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.packages.management.storage.bookkeeper;
+
+import org.apache.pulsar.packages.management.core.PackagesStorage;
+import org.apache.pulsar.packages.management.core.PackagesStorageConfiguration;
+import org.apache.pulsar.packages.management.core.PackagesStorageProvider;
+
+public class BookKeeperPackagesStorageProvider implements PackagesStorageProvider {
+    @Override
+    public PackagesStorage getStorage(PackagesStorageConfiguration config) {
+        return new BookKeeperPackagesStorage(config);
+    }
+}