Skip to content

Commit

Permalink
Setup the initial namespace of the cluster without startup the Pulsar…
Browse files Browse the repository at this point in the history
… broker. (apache#7434)
  • Loading branch information
codelipenghui authored Jul 3, 2020
1 parent 56fe11a commit 35bb235
Show file tree
Hide file tree
Showing 2 changed files with 102 additions and 0 deletions.
3 changes: 3 additions & 0 deletions bin/pulsar
Original file line number Diff line number Diff line change
Expand Up @@ -144,6 +144,7 @@ where command is one of:
initialize-cluster-metadata One-time metadata initialization
initialize-transaction-coordinator-metadata One-time transaction coordinator metadata initialization
initialize-namespace namespace initialization
compact-topic Run compaction against a topic
zookeeper-shell Open a ZK shell client
broker-tool CLI to operate a specific broker
Expand Down Expand Up @@ -340,6 +341,8 @@ elif [ $COMMAND == "initialize-cluster-metadata" ]; then
exec $JAVA $OPTS org.apache.pulsar.PulsarClusterMetadataSetup $@
elif [ $COMMAND == "initialize-transaction-coordinator-metadata" ]; then
exec $JAVA $OPTS org.apache.pulsar.PulsarTransactionCoordinatorMetadataSetup $@
elif [ $COMMAND == "initialize-namespace" ]; then
exec $JAVA $OPTS org.apache.pulsar.PulsarInitialNamespaceSetup $@
elif [ $COMMAND == "zookeeper-shell" ]; then
exec $JAVA $OPTS org.apache.zookeeper.ZooKeeperMain $@
elif [ $COMMAND == "broker-tool" ]; then
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,99 @@
/**
* 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;

import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.zookeeper.ZooKeeper;

import java.util.List;

/**
* Setup the initial namespace of the cluster without startup the Pulsar broker.
*/
public class PulsarInitialNamespaceSetup {

private static class Arguments {

@Parameter(names = { "-c", "--cluster" }, description = "Cluster name", required = true)
private String cluster;

@Parameter(names = { "-cs",
"--configuration-store" }, description = "Configuration Store connection string", required = true)
private String configurationStore;

@Parameter(names = {
"--zookeeper-session-timeout-ms"
}, description = "Local zookeeper session timeout ms")
private int zkSessionTimeoutMillis = 30000;

@Parameter(description = "tenant/namespace", required = true)
private List<String> namespaces;

@Parameter(names = { "-h", "--help" }, description = "Show this help message")
private boolean help = false;

}

public static void main(String[] args) throws Exception {
Arguments arguments = new Arguments();
JCommander jcommander = new JCommander();
try {
jcommander.addObject(arguments);
jcommander.parse(args);
if (arguments.help) {
jcommander.usage();
return;
}
} catch (Exception e) {
jcommander.usage();
throw e;
}

if (arguments.configurationStore == null) {
System.err.println("Configuration store address argument is required (--configuration-store)");
jcommander.usage();
System.exit(1);
}

ZooKeeper configStoreZk = PulsarClusterMetadataSetup
.initZk(arguments.configurationStore, arguments.zkSessionTimeoutMillis);

for (String namespace : arguments.namespaces) {
NamespaceName namespaceName = null;
try {
namespaceName = NamespaceName.get(namespace);
} catch (Exception e) {
System.out.println("Invalid namespace name.");
System.exit(1);
}

// Create system tenant
PulsarClusterMetadataSetup
.createTenantIfAbsent(configStoreZk, namespaceName.getTenant(), arguments.cluster);

// Create system namespace
PulsarClusterMetadataSetup.createNamespaceIfAbsent(configStoreZk, namespaceName,
arguments.cluster);
}

System.out.println("Initial namespace setup success");
}
}

0 comments on commit 35bb235

Please sign in to comment.