Enable bookkeeper table service in pulsar standalone (#1922)

This change is enabling bookkeeper table service component in pulsar standalone. So people can try out stateful functions in pulsar. Signed-off-by: Sijie Guo <sijie@apache.org>
apache · Jun 20, 2018 · da48f42 · da48f42
1 parent c69003d
commit da48f42
Show file tree

Hide file tree

Showing 10 changed files with 134 additions and 28 deletions.
diff --git a/pom.xml b/pom.xml
@@ -121,7 +121,7 @@ flexible messaging model and an intuitive client API.</description>
     <testRealAWS>false</testRealAWS>
     <testRetryCount>1</testRetryCount>
 
-    <bookkeeper.version>4.7.0</bookkeeper.version>
+    <bookkeeper.version>4.7.1</bookkeeper.version>
     <zookeeper.version>3.5.4-beta</zookeeper.version>
     <netty.version>4.1.22.Final</netty.version>
     <storm.version>1.0.5</storm.version>
@@ -280,12 +280,12 @@ flexible messaging model and an intuitive client API.</description>
         <groupId>org.apache.bookkeeper</groupId>
         <artifactId>stream-storage-java-client</artifactId>
         <version>${bookkeeper.version}</version>
-        <exclusions>
-          <exclusion>
-            <groupId>*</groupId>
-            <artifactId>*</artifactId>
-          </exclusion>
-        </exclusions>
+      </dependency>
+
+      <dependency>
+        <groupId>org.apache.bookkeeper</groupId>
+        <artifactId>stream-storage-server</artifactId>
+        <version>${bookkeeper.version}</version>
       </dependency>
 
       <dependency>

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarStandaloneStarter.java
@@ -87,6 +87,12 @@ public class PulsarStandaloneStarter {
     @Parameter(names = {"-fwc", "--functions-worker-conf"}, description = "Configuration file for Functions Worker")
     private String fnWorkerConfigFile = Paths.get("").toAbsolutePath().normalize().toString() + "/conf/functions_worker.yml";
 
+    @Parameter(names = {"-nss", "--no-stream-storage"}, description = "Disable stream storage")
+    private boolean noStreamStorage = false;
+
+    @Parameter(names = { "--stream-storage-port" }, description = "Local bookies stream storage port")
+    private int streamStoragePort = 4181;
+
     @Parameter(names = { "-a", "--advertised-address" }, description = "Standalone broker advertised address")
     private String advertisedAddress = null;
 
@@ -170,8 +176,9 @@ void start() throws Exception {
 
         if (!onlyBroker) {
             // Start LocalBookKeeper
-            bkEnsemble = new LocalBookkeeperEnsemble(numOfBk, zkPort, bkPort, zkDir, bkDir, wipeData, config.getAdvertisedAddress());
-            bkEnsemble.startStandalone();
+            bkEnsemble = new LocalBookkeeperEnsemble(
+                numOfBk, zkPort, bkPort, streamStoragePort, zkDir, bkDir, wipeData, config.getAdvertisedAddress());
+            bkEnsemble.startStandalone(!noStreamStorage);
         }
 
         if (noBroker) {
@@ -189,6 +196,7 @@ void start() throws Exception {
             // worker talks to local broker
             workerConfig.setPulsarServiceUrl("pulsar://127.0.0.1:" + config.getBrokerServicePort());
             workerConfig.setPulsarWebServiceUrl("http://127.0.0.1:" + config.getWebServicePort());
+            workerConfig.setStateStorageServiceUrl("bk://127.0.0.1:" + streamStoragePort);
             String hostname = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(
                 config.getAdvertisedAddress());
             workerConfig.setWorkerHostname(hostname);

diff --git a/pulsar-client-cpp/run-unit-tests.sh b/pulsar-client-cpp/run-unit-tests.sh
@@ -25,11 +25,12 @@ rm -rf ./pulsar-dist
 mkdir pulsar-dist
 tar xfz ../all/target/apache-pulsar*bin.tar.gz  -C pulsar-dist --strip-components 1
 
-PULSAR_STANDALONE_CONF=$PWD/test-conf/standalone.conf pulsar-dist/bin/pulsar standalone --no-functions-worker > broker.log &
+PULSAR_STANDALONE_CONF=$PWD/test-conf/standalone.conf pulsar-dist/bin/pulsar standalone --no-functions-worker --no-stream-storage > broker.log &
 standalone_pid=$!;
 
 PULSAR_STANDALONE_CONF=$PWD/test-conf/standalone-ssl.conf pulsar-dist/bin/pulsar standalone \
               --no-functions-worker \
+              --no-stream-storage \
               --zookeeper-port 2191 --bookkeeper-port 3191 \
               --zookeeper-dir data2/standalone/zookeeper --bookkeeper-dir \
               data2/standalone/bookkeeper > broker-tls.log &

diff --git a/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java b/pulsar-client-tools-test/src/test/java/org/apache/pulsar/admin/cli/CmdFunctionsTest.java
@@ -572,7 +572,7 @@ public void testStateGetter() throws Exception {
             "--namespace", namespace,
             "--name", fnName,
             "--key", "test-key",
-            "--storage-service-url", "127.0.0.1:4181"
+            "--storage-service-url", "bk://127.0.0.1:4181"
         });
 
         assertEquals(

diff --git a/pulsar-client-tools/pom.xml b/pulsar-client-tools/pom.xml
@@ -88,12 +88,6 @@
     <dependency>
       <groupId>org.apache.bookkeeper</groupId>
       <artifactId>stream-storage-java-client</artifactId>
-      <exclusions>
-        <exclusion>
-          <groupId>*</groupId>
-          <artifactId>*</artifactId>
-        </exclusion>
-      </exclusions>
     </dependency>
 
     <!-- functions related dependencies (end) -->

diff --git a/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java b/pulsar-client-tools/src/main/java/org/apache/pulsar/admin/cli/CmdFunctions.java
@@ -46,7 +46,6 @@
 import org.apache.bookkeeper.api.kv.result.KeyValue;
 import org.apache.bookkeeper.clients.StorageClientBuilder;
 import org.apache.bookkeeper.clients.config.StorageClientSettings;
-import org.apache.bookkeeper.clients.utils.NetUtils;
 import org.apache.commons.lang.StringUtils;
 import static org.apache.commons.lang.StringUtils.isNotBlank;
 import static org.apache.commons.lang3.StringUtils.isBlank;
@@ -798,7 +797,7 @@ void runCmd() throws Exception {
 
             try (StorageClient client = StorageClientBuilder.newBuilder()
                  .withSettings(StorageClientSettings.newBuilder()
-                     .addEndpoints(NetUtils.parseEndpoint(stateStorageServiceUrl))
+                     .serviceUri(stateStorageServiceUrl)
                      .clientName("functions-admin")
                      .build())
                  .withNamespace(tableNs)

diff --git a/pulsar-functions/instance/pom.xml b/pulsar-functions/instance/pom.xml
@@ -74,12 +74,6 @@
     <dependency>
       <groupId>org.apache.bookkeeper</groupId>
       <artifactId>stream-storage-java-client</artifactId>
-      <exclusions>
-        <exclusion>
-          <groupId>*</groupId>
-          <artifactId>*</artifactId>
-        </exclusion>
-      </exclusions>
     </dependency>
 
     <dependency>

diff --git a/...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java b/...ons/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -43,7 +43,6 @@
 import org.apache.bookkeeper.clients.config.StorageClientSettings;
 import org.apache.bookkeeper.clients.exceptions.NamespaceNotFoundException;
 import org.apache.bookkeeper.clients.exceptions.StreamNotFoundException;
-import org.apache.bookkeeper.clients.utils.NetUtils;
 import org.apache.bookkeeper.stream.proto.NamespaceConfiguration;
 import org.apache.logging.log4j.ThreadContext;
 import org.apache.logging.log4j.core.LoggerContext;
@@ -262,7 +261,7 @@ private void setupStateTable() throws Exception {
 
         // TODO (sijie): use endpoint for now
         StorageClientSettings settings = StorageClientSettings.newBuilder()
-                .addEndpoints(NetUtils.parseEndpoint(stateStorageServiceUrl))
+                .serviceUri(stateStorageServiceUrl)
                 .clientName("function-" + tableNs + "/" + tableName)
                 .build();
 

diff --git a/pulsar-zookeeper-utils/pom.xml b/pulsar-zookeeper-utils/pom.xml
@@ -44,6 +44,11 @@
       <artifactId>bookkeeper-server</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>org.apache.bookkeeper</groupId>
+      <artifactId>stream-storage-server</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.bookkeeper.stats</groupId>
       <artifactId>prometheus-metrics-provider</artifactId>

diff --git a/...ar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java b/...ar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java
@@ -23,6 +23,7 @@
 
 package org.apache.pulsar.zookeeper;
 
+import static org.apache.bookkeeper.stream.protocol.ProtocolConstants.DEFAULT_STREAM_CONF;
 import static org.apache.commons.io.FileUtils.cleanDirectory;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 
@@ -33,17 +34,35 @@
 import java.io.OutputStream;
 import java.net.InetSocketAddress;
 import java.net.Socket;
+import java.net.URI;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.bookkeeper.bookie.BookieException.InvalidCookieException;
 import org.apache.bookkeeper.bookie.storage.ldb.DbLedgerStorage;
+import org.apache.bookkeeper.clients.StorageClientBuilder;
+import org.apache.bookkeeper.clients.admin.StorageAdminClient;
+import org.apache.bookkeeper.clients.config.StorageClientSettings;
+import org.apache.bookkeeper.clients.exceptions.ClientException;
+import org.apache.bookkeeper.clients.exceptions.NamespaceExistsException;
+import org.apache.bookkeeper.clients.exceptions.NamespaceNotFoundException;
+import org.apache.bookkeeper.common.concurrent.FutureUtils;
+import org.apache.bookkeeper.common.util.Backoff;
+import org.apache.bookkeeper.common.util.Backoff.Jitter.Type;
 import org.apache.bookkeeper.conf.ServerConfiguration;
 import org.apache.bookkeeper.proto.BookieServer;
+import org.apache.bookkeeper.server.conf.BookieConfiguration;
 import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.stream.proto.NamespaceConfiguration;
+import org.apache.bookkeeper.stream.proto.NamespaceProperties;
+import org.apache.bookkeeper.stream.server.StreamStorageLifecycleComponent;
+import org.apache.bookkeeper.stream.storage.api.cluster.ClusterInitializer;
+import org.apache.bookkeeper.stream.storage.impl.cluster.ZkClusterInitializer;
 import org.apache.bookkeeper.util.MathUtils;
+import org.apache.commons.configuration.CompositeConfiguration;
+import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.WatchedEvent;
@@ -69,15 +88,27 @@ public LocalBookkeeperEnsemble(int numberOfBookies, int zkPort, int bkBasePort)
 
     public LocalBookkeeperEnsemble(int numberOfBookies, int zkPort, int bkBasePort, String zkDataDirName,
             String bkDataDirName, boolean clearOldData) {
-        this(numberOfBookies, zkPort, bkBasePort, zkDataDirName, bkDataDirName, clearOldData, null);
+        this(numberOfBookies, zkPort, bkBasePort, 4181, zkDataDirName, bkDataDirName, clearOldData, null);
     }
 
     public LocalBookkeeperEnsemble(int numberOfBookies, int zkPort, int bkBasePort, String zkDataDirName,
             String bkDataDirName, boolean clearOldData, String advertisedAddress) {
+        this(numberOfBookies, zkPort, bkBasePort, 4181, zkDataDirName, bkDataDirName, clearOldData, advertisedAddress);
+    }
+
+    public LocalBookkeeperEnsemble(int numberOfBookies,
+                                   int zkPort,
+                                   int bkBasePort,
+                                   int streamStoragePort,
+                                   String zkDataDirName,
+                                   String bkDataDirName,
+                                   boolean clearOldData,
+                                   String advertisedAddress) {
         this.numberOfBookies = numberOfBookies;
         this.HOSTPORT = "127.0.0.1:" + zkPort;
         this.ZooKeeperDefaultPort = zkPort;
         this.initialPort = bkBasePort;
+        this.streamStoragePort = streamStoragePort;
         this.zkDataDirName = zkDataDirName;
         this.bkDataDirName = bkDataDirName;
         this.clearOldData = clearOldData;
@@ -100,6 +131,10 @@ public LocalBookkeeperEnsemble(int numberOfBookies, int zkPort, int bkBasePort,
     ServerConfiguration bsConfs[];
     Integer initialPort = 5000;
 
+    // Stream/Table Storage
+    StreamStorageLifecycleComponent streamStorage;
+    Integer streamStoragePort = 4181;
+
     /**
      * @param args
      */
@@ -213,6 +248,65 @@ private void runBookies(ServerConfiguration baseConf) throws Exception {
         }
     }
 
+    private void runStreamStorage(CompositeConfiguration conf) throws Exception {
+        String zkServers = "127.0.0.1:" + ZooKeeperDefaultPort;
+        String metadataServiceUriStr = "zk://" + zkServers + "/ledgers";
+        URI metadataServiceUri = URI.create(metadataServiceUriStr);
+
+        // zookeeper servers
+        conf.setProperty("metadataServiceUri", metadataServiceUriStr);
+        // dlog settings
+        conf.setProperty("dlog.bkcEnsembleSize", 1);
+        conf.setProperty("dlog.bkcWriteQuorumSize", 1);
+        conf.setProperty("dlog.bkcAckQuorumSize", 1);
+        // stream storage port
+        conf.setProperty("storageserver.grpc.port", streamStoragePort);
+
+        // initialize the stream storage metadata
+        ClusterInitializer initializer = new ZkClusterInitializer(zkServers);
+        initializer.initializeCluster(metadataServiceUri, 2);
+
+        // load the stream storage component
+        ServerConfiguration serverConf = new ServerConfiguration();
+        serverConf.loadConf(conf);
+        BookieConfiguration bkConf = new BookieConfiguration(serverConf);
+
+        this.streamStorage = new StreamStorageLifecycleComponent(bkConf, NullStatsLogger.INSTANCE);
+        this.streamStorage.start();
+        LOG.debug("Local BK stream storage started (port: {})", streamStoragePort);
+
+        // create a default namespace
+        try (StorageAdminClient admin = StorageClientBuilder.newBuilder()
+             .withSettings(StorageClientSettings.newBuilder()
+                 .serviceUri("bk://localhost:4181")
+                 .backoffPolicy(Backoff.Jitter.of(
+                     Type.EXPONENTIAL,
+                     1000,
+                     10000,
+                     30
+                 ))
+                 .build())
+            .buildAdmin()) {
+
+            try {
+                NamespaceProperties ns = FutureUtils.result(admin.getNamespace("default"));
+                LOG.info("'default' namespace for table service : {}", ns);
+            } catch (NamespaceNotFoundException nnfe) {
+                LOG.info("Creating default namespace");
+                try {
+                    NamespaceProperties ns =
+                        FutureUtils.result(admin.createNamespace("default", NamespaceConfiguration.newBuilder()
+                            .setDefaultStreamConf(DEFAULT_STREAM_CONF)
+                            .build()));
+                    LOG.info("Successfully created 'default' namespace :\n{}", ns);
+                } catch (NamespaceExistsException nee) {
+                    // namespace already exists
+                    LOG.warn("Namespace 'default' already existed.");
+                }
+            }
+        }
+    }
+
     public void start() throws Exception {
         LOG.debug("Local ZK/BK starting ...");
         ServerConfiguration conf = new ServerConfiguration();
@@ -232,6 +326,10 @@ public void start() throws Exception {
     }
 
     public void startStandalone() throws Exception {
+        startStandalone(false);
+    }
+
+    public void startStandalone(boolean enableStreamStorage) throws Exception {
         LOG.debug("Local ZK/BK starting ...");
         ServerConfiguration conf = new ServerConfiguration();
         conf.setLedgerManagerFactoryClassName("org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory");
@@ -248,9 +346,17 @@ public void startStandalone() throws Exception {
         runZookeeper(1000);
         initializeZookeper();
         runBookies(conf);
+        if (enableStreamStorage) {
+            runStreamStorage(new CompositeConfiguration());
+        }
     }
 
     public void stop() throws Exception {
+        if (null != streamStorage) {
+            LOG.debug("Local bk stream storage stopping ...");
+            streamStorage.close();
+        }
+
         LOG.debug("Local ZK/BK stopping ...");
         for (BookieServer bookie : bs) {
             bookie.shutdown();