You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2018/01/04 06:09:38 UTC
[GitHub] merlimat closed pull request #1023: Refine broker starter command
merlimat closed pull request #1023: Refine broker starter command
URL: https://github.com/apache/incubator-pulsar/pull/1023
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/bin/pulsar b/bin/pulsar
index 1bfd99528..927a15d58 100755
--- a/bin/pulsar
+++ b/bin/pulsar
@@ -196,7 +196,7 @@ OPTS="$OPTS -Dpulsar.log.dir=$PULSAR_LOG_DIR"
cd "$PULSAR_HOME"
if [ $COMMAND == "broker" ]; then
PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"pulsar-broker.log"}
- exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarBrokerStarter $PULSAR_BROKER_CONF $@
+ exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.pulsar.PulsarBrokerStarter --broker-conf $PULSAR_BROKER_CONF $@
elif [ $COMMAND == "bookie" ]; then
PULSAR_LOG_FILE=${PULSAR_LOG_FILE:-"bookkeeper.log"}
exec $JAVA $OPTS -Dpulsar.log.file=$PULSAR_LOG_FILE org.apache.bookkeeper.proto.BookieServer --conf $PULSAR_BOOKKEEPER_CONF $@
diff --git a/conf/broker.conf b/conf/broker.conf
index f9f802b90..af05b567c 100644
--- a/conf/broker.conf
+++ b/conf/broker.conf
@@ -159,6 +159,12 @@ enablePersistentTopics=true
# Enable broker to load non-persistent topics
enableNonPersistentTopics=true
+# Enable to run bookie along with broker
+enableRunBookieTogether=false
+
+// Enable to run bookie autorecovery along with broker
+enableRunBookieAutoRecoveryTogether=false
+
### --- Authentication --- ###
# Enable TLS
diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
index f0aae8247..a638fd4fb 100644
--- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
+++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java
@@ -160,6 +160,12 @@
// Enable broker to load non-persistent topics
private boolean enableNonPersistentTopics = true;
+ // Enable to run bookie along with broker
+ private boolean enableRunBookieTogether = false;
+
+ // Enable to run bookie autorecovery along with broker
+ private boolean enableRunBookieAutoRecoveryTogether = false;
+
/***** --- TLS --- ****/
// Enable TLS
private boolean tlsEnabled = false;
@@ -291,7 +297,7 @@
private int loadBalancerReportUpdateMaxIntervalMinutes = 15;
// Frequency of report to collect
private int loadBalancerHostUsageCheckIntervalMinutes = 1;
- // Enable/disable automatic bundle unloading for load-shedding
+ // Enable/disable automatic bundle unloading for load-shedding
@FieldContext(dynamic = true)
private boolean loadBalancerSheddingEnabled = true;
// Load shedding interval. Broker periodically checks whether some traffic should be offload from some over-loaded
@@ -678,6 +684,22 @@ public void setEnableNonPersistentTopics(boolean enableNonPersistentTopics) {
this.enableNonPersistentTopics = enableNonPersistentTopics;
}
+ public boolean isEnableRunBookieTogether() {
+ return enableRunBookieTogether;
+ }
+
+ public void setEnableRunBookieTogether(boolean enableRunBookieTogether) {
+ this.enableRunBookieTogether = enableRunBookieTogether;
+ }
+
+ public boolean isEnableRunBookieAutoRecoveryTogether() {
+ return enableRunBookieAutoRecoveryTogether;
+ }
+
+ public void setEnableRunBookieAutoRecoveryTogether(boolean enableRunBookieAutoRecoveryTogether) {
+ this.enableRunBookieAutoRecoveryTogether = enableRunBookieAutoRecoveryTogether;
+ }
+
public boolean isTlsEnabled() {
return tlsEnabled;
}
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
index ccdf6547e..40f8d59ca 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/PulsarBrokerStarter.java
@@ -28,9 +28,11 @@
import com.beust.jcommander.JCommander;
import com.beust.jcommander.Parameter;
import com.ea.agentloader.AgentLoader;
+import com.google.common.annotations.VisibleForTesting;
import java.io.File;
import java.io.FileInputStream;
import java.net.MalformedURLException;
+import java.nio.file.Paths;
import java.util.Arrays;
import org.apache.bookkeeper.conf.ServerConfiguration;
import org.apache.bookkeeper.proto.BookieServer;
@@ -38,7 +40,6 @@
import org.apache.bookkeeper.stats.StatsProvider;
import org.apache.bookkeeper.util.ReflectionUtils;
import org.apache.commons.configuration.ConfigurationException;
-import org.apache.pulsar.broker.PulsarServerException;
import org.apache.pulsar.broker.PulsarService;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.aspectj.weaver.loadtime.Agent;
@@ -57,16 +58,19 @@ private static ServiceConfiguration loadConfig(String configFile) throws Excepti
return config;
}
- private static class BookieArguments {
+ @VisibleForTesting
+ private static class StarterArguments {
+ @Parameter(names = {"-c", "--broker-conf"}, description = "Configuration file for Broker")
+ private String brokerConfigFile = Paths.get("").toAbsolutePath().normalize().toString() + "/conf/broker.conf";
- @Parameter(names = {"-rb", "--run-bookie"}, description = "Run Bookie together with broker")
+ @Parameter(names = {"-rb", "--run-bookie"}, description = "Run Bookie together with Broker")
private boolean runBookie = false;
@Parameter(names = {"-ra", "--run-bookie-autorecovery"}, description = "Run Bookie Autorecovery together with broker")
private boolean runBookieAutoRecovery = false;
@Parameter(names = {"-bc", "--bookie-conf"}, description = "Configuration file for Bookie")
- private String bookieConfigFile;
+ private String bookieConfigFile = Paths.get("").toAbsolutePath().normalize().toString() + "/conf/bookkeeper.conf";
@Parameter(names = {"-h", "--help"}, description = "Show this help message")
private boolean help = false;
@@ -88,34 +92,62 @@ private static ServerConfiguration readBookieConfFile(String bookieConfigFile) t
return bookieConf;
}
- private static class PulsarBookieStarter {
+ private static boolean argsContains(String[] args, String arg) {
+ return Arrays.asList(args).contains(arg);
+ }
+
+ private static class BrokerStarter {
+ private final ServiceConfiguration brokerConfig;
+ private final PulsarService pulsarService;
private final BookieServer bookieServer;
private final AutoRecoveryMain autoRecoveryMain;
private final StatsProvider bookieStatsProvider;
private final ServerConfiguration bookieConfig;
- PulsarBookieStarter(String[] args) throws Exception{
- BookieArguments bookieArguments = new BookieArguments();
- JCommander jcommander = new JCommander(bookieArguments);
- jcommander.setProgramName("PulsarBrokerStarter <broker.conf>");
+ BrokerStarter(String[] args) throws Exception{
+ StarterArguments starterArguments = new StarterArguments();
+ JCommander jcommander = new JCommander(starterArguments);
+ jcommander.setProgramName("PulsarBrokerStarter");
- // parse args by jcommander
+ // parse args by JCommander
jcommander.parse(args);
- if (bookieArguments.help) {
+ if (starterArguments.help) {
jcommander.usage();
System.exit(-1);
}
- if ((bookieArguments.runBookie || bookieArguments.runBookieAutoRecovery)
- && isBlank(bookieArguments.bookieConfigFile)) {
+
+ // init broker config and pulsar service
+ if (isBlank(starterArguments.brokerConfigFile)) {
+ jcommander.usage();
+ throw new IllegalArgumentException("Need to specify a configuration file for broker");
+ } else {
+ brokerConfig = loadConfig(starterArguments.brokerConfigFile);
+ pulsarService = new PulsarService(brokerConfig);
+ }
+
+ // if no argument to run bookie in cmd line, read from pulsar config
+ if (!argsContains(args, "-rb") && !argsContains(args, "--run-bookie")) {
+ checkState(starterArguments.runBookie == false,
+ "runBookie should be false if has no argument specified");
+ starterArguments.runBookie = brokerConfig.isEnableRunBookieTogether();
+ }
+ if (!argsContains(args, "-ra") && !argsContains(args, "--run-bookie-autorecovery")) {
+ checkState(starterArguments.runBookieAutoRecovery == false,
+ "runBookieAutoRecovery should be false if has no argument specified");
+ starterArguments.runBookieAutoRecovery = brokerConfig.isEnableRunBookieAutoRecoveryTogether();
+ }
+
+ if ((starterArguments.runBookie || starterArguments.runBookieAutoRecovery)
+ && isBlank(starterArguments.bookieConfigFile)) {
jcommander.usage();
throw new IllegalArgumentException("No configuration file for Bookie");
}
// init stats provider
- if (bookieArguments.runBookie || bookieArguments.runBookieAutoRecovery) {
- checkState(isNotBlank(bookieArguments.bookieConfigFile),
+ if (starterArguments.runBookie || starterArguments.runBookieAutoRecovery) {
+ checkState(isNotBlank(starterArguments.bookieConfigFile),
"No configuration file for Bookie");
- bookieConfig = readBookieConfFile(bookieArguments.bookieConfigFile);
+ bookieConfig = readBookieConfFile(starterArguments.bookieConfigFile);
Class<? extends StatsProvider> statsProviderClass = bookieConfig.getStatsProviderClass();
bookieStatsProvider = ReflectionUtils.newInstance(statsProviderClass);
} else {
@@ -124,7 +156,7 @@ private static ServerConfiguration readBookieConfFile(String bookieConfigFile) t
}
// init bookie server
- if (bookieArguments.runBookie) {
+ if (starterArguments.runBookie) {
checkNotNull(bookieConfig, "No ServerConfiguration for Bookie");
checkNotNull(bookieStatsProvider, "No Stats Provider for Bookie");
bookieServer = new BookieServer(bookieConfig, bookieStatsProvider.getStatsLogger(""));
@@ -133,7 +165,7 @@ private static ServerConfiguration readBookieConfFile(String bookieConfigFile) t
}
// init bookie AutorecoveryMain
- if (bookieArguments.runBookieAutoRecovery) {
+ if (starterArguments.runBookieAutoRecovery) {
checkNotNull(bookieConfig, "No ServerConfiguration for Bookie Autorecovery");
autoRecoveryMain = new AutoRecoveryMain(bookieConfig);
} else {
@@ -154,9 +186,14 @@ public void start() throws Exception {
autoRecoveryMain.start();
log.info("started bookie autoRecoveryMain.");
}
+
+ pulsarService.start();
+ log.info("PulsarService started.");
}
public void join() throws InterruptedException {
+ pulsarService.waitUntilClosed();
+
if (bookieServer != null) {
bookieServer.join();
}
@@ -166,59 +203,48 @@ public void join() throws InterruptedException {
}
public void shutdown() {
+ pulsarService.getShutdownService().run();
+ log.info("Shut down broker service successfully.");
+
if (bookieStatsProvider != null) {
bookieStatsProvider.stop();
+ log.info("Shut down bookieStatsProvider successfully.");
}
if (bookieServer != null) {
bookieServer.shutdown();
+ log.info("Shut down bookieServer successfully.");
}
if (autoRecoveryMain != null) {
autoRecoveryMain.shutdown();
+ log.info("Shut down autoRecoveryMain successfully.");
}
}
}
public static void main(String[] args) throws Exception {
- if (args.length < 1) {
- throw new IllegalArgumentException("Need to specify a configuration file");
- }
-
Thread.setDefaultUncaughtExceptionHandler((thread, exception) -> {
log.error("Uncaught exception in thread {}: {}", thread.getName(), exception.getMessage(), exception);
});
- String configFile = args[0];
- ServiceConfiguration config = loadConfig(configFile);
-
// load aspectj-weaver agent for instrumentation
AgentLoader.loadAgentClass(Agent.class.getName(), null);
- PulsarBookieStarter bookieStarter = new PulsarBookieStarter(Arrays.copyOfRange(args, 1, args.length));
- bookieStarter.start();
-
- @SuppressWarnings("resource")
- final PulsarService service = new PulsarService(config);
+ BrokerStarter starter = new BrokerStarter(args);
Runtime.getRuntime().addShutdownHook(
new Thread(() -> {
- service.getShutdownService().run();
- log.info("Shut down broker service successfully.");
- bookieStarter.shutdown();
+ starter.shutdown();
})
);
try {
- service.start();
- log.info("PulsarService started");
- } catch (PulsarServerException e) {
+ starter.start();
+ } catch (Exception e) {
log.error("Failed to start pulsar service.", e);
-
Runtime.getRuntime().halt(1);
}
- service.waitUntilClosed();
-
- bookieStarter.join();
+ starter.join();
}
private static final Logger log = LoggerFactory.getLogger(PulsarBrokerStarter.class);
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/PulsarBrokerStarterTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/PulsarBrokerStarterTest.java
index 31290e42d..0cd9e879c 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/PulsarBrokerStarterTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/PulsarBrokerStarterTest.java
@@ -20,6 +20,7 @@
import static org.testng.Assert.fail;
+import com.google.common.collect.Sets;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
@@ -28,15 +29,11 @@
import java.io.PrintWriter;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
-
import org.apache.bookkeeper.bookie.LedgerDirsManager;
-import org.apache.pulsar.PulsarBrokerStarter;
import org.apache.pulsar.broker.ServiceConfiguration;
import org.testng.Assert;
import org.testng.annotations.Test;
-import com.google.common.collect.Sets;
-
/**
* @version $Revision$<br>
* Created on Sep 6, 2012
@@ -77,6 +74,7 @@ private File createValidBrokerConfigFile() throws FileNotFoundException {
printWriter.println("replicationProducerQueueSize=50");
printWriter.println("bookkeeperClientTimeoutInSeconds=12345");
printWriter.println("bookkeeperClientSpeculativeReadTimeoutInMillis=3000");
+ printWriter.println("enableRunBookieTogether=true");
printWriter.close();
testConfigFile.deleteOnExit();
@@ -266,19 +264,18 @@ public void testGlobalZooKeeperConfig() throws SecurityException, NoSuchMethodEx
}
/**
- * Verifies that the main throws {@link IllegalArgumentException} when no argument is given.
+ * Verifies that the main throws {@link FileNotFoundException} when no argument is given.
*/
@Test
public void testMainWithNoArgument() throws Exception {
try {
PulsarBrokerStarter.main(new String[0]);
- Assert.fail("No argument to main should've raised IllegalArgumentException!");
- } catch (IllegalArgumentException e) {
+ Assert.fail("No argument to main should've raised FileNotFoundException for no broker config!");
+ } catch (FileNotFoundException e) {
// code should reach here.
}
}
-
/**
* Verifies that the main throws {@link IllegalArgumentException}
* when no config file for bookie and bookie auto recovery is given.
@@ -287,11 +284,12 @@ public void testMainWithNoArgument() throws Exception {
public void testMainRunBookieAndAutoRecoveryNoConfig() throws Exception {
try {
File testConfigFile = createValidBrokerConfigFile();
- String[] args = {testConfigFile.getAbsolutePath(), "-rb", "-ra"};
+ String[] args = {"-c", testConfigFile.getAbsolutePath(), "-rb", "-ra", "-bc", ""};
PulsarBrokerStarter.main(args);
Assert.fail("No Config file for bookie auto recovery should've raised IllegalArgumentException!");
} catch (IllegalArgumentException e) {
// code should reach here.
+ e.printStackTrace();
Assert.assertEquals(e.getMessage(), "No configuration file for Bookie");
}
}
@@ -304,7 +302,7 @@ public void testMainRunBookieAndAutoRecoveryNoConfig() throws Exception {
public void testMainRunBookieRecoveryNoConfig() throws Exception {
try {
File testConfigFile = createValidBrokerConfigFile();
- String[] args = {testConfigFile.getAbsolutePath(), "-ra"};
+ String[] args = {"-c", testConfigFile.getAbsolutePath(), "-ra", "-bc", ""};
PulsarBrokerStarter.main(args);
Assert.fail("No Config file for bookie auto recovery should've raised IllegalArgumentException!");
} catch (IllegalArgumentException e) {
@@ -320,7 +318,7 @@ public void testMainRunBookieRecoveryNoConfig() throws Exception {
public void testMainRunBookieNoConfig() throws Exception {
try {
File testConfigFile = createValidBrokerConfigFile();
- String[] args = {testConfigFile.getAbsolutePath(), "-rb"};
+ String[] args = {"-c", testConfigFile.getAbsolutePath(), "-rb", "-bc", ""};
PulsarBrokerStarter.main(args);
Assert.fail("No Config file for bookie should've raised IllegalArgumentException!");
} catch (IllegalArgumentException e) {
@@ -330,13 +328,14 @@ public void testMainRunBookieNoConfig() throws Exception {
}
/**
- * Verifies that the main throws {@link IllegalArgumentException} when no config file for bookie is given.
+ * Verifies that the main throws {@link IllegalArgumentException} when malformed config file for bookie is given.
*/
@Test
public void testMainRunBookieEmptyConfig() throws Exception {
try {
File testConfigFile = createValidBrokerConfigFile();
- String[] args = {testConfigFile.getAbsolutePath(), "-ra", "-rb", "-bc", testConfigFile.getAbsolutePath()};
+ String[] args = {"-c", testConfigFile.getAbsolutePath(),
+ "-ra", "-rb", "-bc", testConfigFile.getAbsolutePath()};
PulsarBrokerStarter.main(args);
Assert.fail("Effectively empty config file for bookie should've raised NoWritableLedgerDirException!");
} catch (LedgerDirsManager.NoWritableLedgerDirException e) {
@@ -344,4 +343,19 @@ public void testMainRunBookieEmptyConfig() throws Exception {
// Since empty config file will have empty ledgerDirs, it should raise exception when bookie init.
}
}
+
+ /**
+ * Verifies that the main throws {@link IllegalArgumentException} when no config file for bookie .
+ */
+ @Test
+ public void testMainEnableRunBookieThroughBrokerConfig() throws Exception {
+ try {
+ File testConfigFile = createValidBrokerConfigFile();
+ String[] args = {"-c", testConfigFile.getAbsolutePath()};
+ PulsarBrokerStarter.main(args);
+ Assert.fail("No argument to main should've raised IllegalArgumentException for no bookie config!");
+ } catch (IllegalArgumentException e) {
+ // code should reach here.
+ }
+ }
}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services