This is an automated email from the ASF dual-hosted git repository.
sijie pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new da48f42 Enable bookkeeper table service in pulsar standalone (#1922)
da48f42 is described below
commit da48f42245b572cc283913f937da476810e03fca
Author: Sijie Guo <[email protected]>
AuthorDate: Wed Jun 20 09:16:06 2018 -0700
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 <[email protected]>
---
pom.xml | 14 +--
.../org/apache/pulsar/PulsarStandaloneStarter.java | 12 ++-
pulsar-client-cpp/run-unit-tests.sh | 3 +-
.../apache/pulsar/admin/cli/CmdFunctionsTest.java | 2 +-
pulsar-client-tools/pom.xml | 6 --
.../org/apache/pulsar/admin/cli/CmdFunctions.java | 3 +-
pulsar-functions/instance/pom.xml | 6 --
.../functions/instance/JavaInstanceRunnable.java | 3 +-
pulsar-zookeeper-utils/pom.xml | 5 +
.../pulsar/zookeeper/LocalBookkeeperEnsemble.java | 108 ++++++++++++++++++++-
10 files changed, 134 insertions(+), 28 deletions(-)
diff --git a/pom.xml b/pom.xml
index cbfffd3..9ae436b 100644
--- 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
index a6a027b..d75d1d9 100644
--- 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 @@ public class PulsarStandaloneStarter {
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 @@ public class PulsarStandaloneStarter {
// 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
index 466eae0..b0bdcd1 100755
--- 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
index e44562a..52b1d26 100644
---
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 class CmdFunctionsTest {
"--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
index 8f94210..c83378c 100644
--- 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
index 67b1377..1dbe50b 100644
---
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.Table;
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 @@ public class CmdFunctions extends CmdBase {
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
index 11cefa5..8b0f763 100644
--- 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/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
index b79d508..c68910b 100644
---
a/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
+++
b/pulsar-functions/instance/src/main/java/org/apache/pulsar/functions/instance/JavaInstanceRunnable.java
@@ -43,7 +43,6 @@ import org.apache.bookkeeper.clients.admin.StorageAdminClient;
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 @@ public class JavaInstanceRunnable implements AutoCloseable,
Runnable {
// 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
index 2e40ce1..ab1ec0c 100644
--- a/pulsar-zookeeper-utils/pom.xml
+++ b/pulsar-zookeeper-utils/pom.xml
@@ -45,6 +45,11 @@
</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>
</dependency>
diff --git
a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java
b/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java
index 6879e0d..9abca4c 100644
---
a/pulsar-zookeeper-utils/src/main/java/org/apache/pulsar/zookeeper/LocalBookkeeperEnsemble.java
+++
b/pulsar-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,6 +34,7 @@ import java.io.InputStreamReader;
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;
@@ -40,10 +42,27 @@ 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 class LocalBookkeeperEnsemble {
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 class LocalBookkeeperEnsemble {
ServerConfiguration bsConfs[];
Integer initialPort = 5000;
+ // Stream/Table Storage
+ StreamStorageLifecycleComponent streamStorage;
+ Integer streamStoragePort = 4181;
+
/**
* @param args
*/
@@ -213,6 +248,65 @@ public class LocalBookkeeperEnsemble {
}
}
+ 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 class LocalBookkeeperEnsemble {
}
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 class LocalBookkeeperEnsemble {
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();
