Use distributedlog-core-shaded in pulsar worker (apache#257)

iEverX · Mar 4, 2018 · 981e472 · 981e472
1 parent 68e1f3c
commit 981e472
Show file tree

Hide file tree

Showing 6 changed files with 16 additions and 9 deletions.
diff --git a/bin/pulsar b/bin/pulsar
@@ -220,7 +220,7 @@ elif [ $COMMAND == "websocket" ]; then
     exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.websocket.service.WebSocketServiceStarter $PULSAR_WEBSOCKET_CONF $@
 elif [ $COMMAND == "standalone" ]; then
     PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-standalone.log"}
-    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE -Dzookeeper.4lw.commands.whitelist='*' org.apache.pulsar.PulsarStandaloneStarter --config $PULSAR_STANDALONE_CONF $@
+    exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarStandaloneStarter --config $PULSAR_STANDALONE_CONF $@
 elif [ $COMMAND == "initialize-cluster-metadata" ]; then
     exec $JAVA $OPTS org.apache.pulsar.PulsarClusterMetadataSetup $@
 elif [ $COMMAND == "zookeeper-shell" ]; then

diff --git a/pom.xml b/pom.xml
@@ -114,7 +114,7 @@ flexible messaging model and an intuitive client API.</description>
     <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding>
 
     <bookkeeper.version>4.7.0-SNAPSHOT</bookkeeper.version>
-    <zookeeper.version>3.5.3-beta</zookeeper.version>
+    <zookeeper.version>3.4.10</zookeeper.version>
     <netty.version>4.1.21.Final</netty.version>
     <storm.version>1.0.5</storm.version>
     <jetty.version>9.3.11.v20160721</jetty.version>

diff --git a/pulsar-functions/dist/pom.xml b/pulsar-functions/dist/pom.xml
@@ -54,6 +54,11 @@
       <scope>provided</scope>
     </dependency>
 
+    <dependency>
+      <groupId>io.netty</groupId>
+      <artifactId>netty-all</artifactId>
+    </dependency>
+
     <dependency>
       <groupId>org.apache.logging.log4j</groupId>
       <artifactId>log4j-slf4j-impl</artifactId>

diff --git a/pulsar-functions/worker/pom.xml b/pulsar-functions/worker/pom.xml
@@ -120,9 +120,10 @@
       <version>1.14.6</version>
     </dependency>
 
+    <!-- use shaded dependency util pulsar bump zookeeper version to 3.5 -->
     <dependency>
       <groupId>org.apache.distributedlog</groupId>
-      <artifactId>distributedlog-core</artifactId>
+      <artifactId>distributedlog-core-shaded</artifactId>
       <version>${bookkeeper.version}</version>
     </dependency>
   </dependencies>

diff --git a/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java b/pulsar-functions/worker/src/main/java/org/apache/pulsar/functions/worker/Utils.java
@@ -18,6 +18,7 @@
  */
 package org.apache.pulsar.functions.worker;
 
+import dlshade.org.apache.zookeeper.KeeperException.Code;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.distributedlog.AppendOnlyStreamWriter;
 import org.apache.distributedlog.DistributedLogConfiguration;
@@ -43,7 +44,6 @@
 import java.net.URI;
 import java.net.URL;
 import java.util.UUID;
-import org.apache.zookeeper.KeeperException.Code;
 
 @Slf4j
 public final class Utils {
@@ -141,7 +141,7 @@ public static void downloadFromBookkeeper(Namespace namespace,
     public static DistributedLogConfiguration getDlogConf(WorkerConfig workerConfig) {
         int numReplicas = workerConfig.getNumFunctionPackageReplicas();
 
-        return new DistributedLogConfiguration()
+        DistributedLogConfiguration conf = new DistributedLogConfiguration()
                 .setWriteLockEnabled(false)
                 .setOutputBufferSize(256 * 1024)                  // 256k
                 .setPeriodicFlushFrequencyMilliSeconds(0)         // disable periodical flush
@@ -154,6 +154,9 @@ public static DistributedLogConfiguration getDlogConf(WorkerConfig workerConfig)
                 .setWriteQuorumSize(numReplicas)
                 .setAckQuorumSize(numReplicas)
                 .setUseDaemonThread(true);
+        conf.setProperty("bkc.allowShadedLedgerManagerFactoryClass", true);
+        conf.setProperty("bkc.shadedLedgerManagerFactoryClassPrefix", "dlshade.");
+        return conf;
     }
 
     public static URI initializeDlogNamespace(String zkServers, String ledgersRootPath) throws IOException {

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
@@ -212,8 +212,7 @@ public void start() throws Exception {
         ServerConfiguration conf = new ServerConfiguration();
         conf.setLedgerManagerFactoryClassName("org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory");
         // Use minimal configuration requiring less memory for unit tests
-        // conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
-        conf.setLedgerStorageClass(SortedLedgerStorage.class.getName());
+        conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
         conf.setProperty("dbStorage_writeCacheMaxSizeMb", 2);
         conf.setProperty("dbStorage_readAheadCacheMaxSizeMb", 1);
         conf.setProperty("dbStorage_rocksDB_writeBufferSizeMB", 1);
@@ -230,8 +229,7 @@ public void startStandalone() throws Exception {
         LOG.debug("Local ZK/BK starting ...");
         ServerConfiguration conf = new ServerConfiguration();
         conf.setLedgerManagerFactoryClassName("org.apache.bookkeeper.meta.HierarchicalLedgerManagerFactory");
-        // conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
-        conf.setLedgerStorageClass(SortedLedgerStorage.class.getName());
+        conf.setLedgerStorageClass(DbLedgerStorage.class.getName());
         conf.setDiskUsageThreshold(0.99999f);
         conf.setDiskUsageWarnThreshold(0.9999f);
         conf.setProperty("dbStorage_writeCacheMaxSizeMb", 256);