helix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From j...@apache.org
Subject [2/2] helix git commit: Start participants for multiple in one JVM
Date Wed, 11 Apr 2018 20:53:16 GMT
Start participants for multiple in one JVM


Project: http://git-wip-us.apache.org/repos/asf/helix/repo
Commit: http://git-wip-us.apache.org/repos/asf/helix/commit/4b50445d
Tree: http://git-wip-us.apache.org/repos/asf/helix/tree/4b50445d
Diff: http://git-wip-us.apache.org/repos/asf/helix/diff/4b50445d

Branch: refs/heads/master
Commit: 4b50445d89182d5ca48d4557c649a64b136e9f79
Parents: 4a76f03
Author: Junkai Xue <jxue@linkedin.com>
Authored: Mon Apr 9 12:28:35 2018 -0700
Committer: Junkai Xue <jxue@linkedin.com>
Committed: Wed Apr 11 13:52:26 2018 -0700

----------------------------------------------------------------------
 helix-core/pom.xml                              |   4 +
 ...rResourceOnlineOfflineStateModelFactory.java |  77 +++++++
 .../SegmentOnlineOfflineStateModelFactory.java  |  53 +++++
 .../tools/commandtools/ExampleParticipant.java  | 207 +++++++++++++++++++
 4 files changed, 341 insertions(+)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/helix/blob/4b50445d/helix-core/pom.xml
----------------------------------------------------------------------
diff --git a/helix-core/pom.xml b/helix-core/pom.xml
index b6a2539..a29243e 100644
--- a/helix-core/pom.xml
+++ b/helix-core/pom.xml
@@ -220,6 +220,10 @@ under the License.
               <name>start-helix-participant</name>
             </program>
             <program>
+              <mainClass>org.apache.helix.tools.commandtools.ExampleParticipant</mainClass>
+              <name>start-participants</name>
+            </program>
+            <program>
               <mainClass>org.apache.helix.tools.LocalZKServer</mainClass>
               <name>start-standalone-zookeeper</name>
             </program>

http://git-wip-us.apache.org/repos/asf/helix/blob/4b50445d/helix-core/src/main/java/org/apache/helix/examples/BrokerResourceOnlineOfflineStateModelFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/examples/BrokerResourceOnlineOfflineStateModelFactory.java
b/helix-core/src/main/java/org/apache/helix/examples/BrokerResourceOnlineOfflineStateModelFactory.java
new file mode 100644
index 0000000..0b1b81f
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/examples/BrokerResourceOnlineOfflineStateModelFactory.java
@@ -0,0 +1,77 @@
+package org.apache.helix.examples;
+
+import org.apache.helix.NotificationContext;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.Transition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class BrokerResourceOnlineOfflineStateModelFactory extends StateModelFactory<StateModel>
{
+  private static final Logger LOGGER = LoggerFactory.getLogger(BrokerResourceOnlineOfflineStateModelFactory.class);
+
+  public BrokerResourceOnlineOfflineStateModelFactory() {
+
+  }
+
+  public static String getStateModelDef() {
+    return "BrokerResourceOnlineOfflineStateModel";
+  }
+
+  public StateModel createNewStateModel(String resourceName) {
+    return new BrokerResourceOnlineOfflineStateModelFactory.BrokerResourceOnlineOfflineStateModel();
+  }
+
+  @StateModelInfo(
+      states = {"{'OFFLINE','ONLINE', 'DROPPED'}"},
+      initialState = "OFFLINE"
+  )
+  public class BrokerResourceOnlineOfflineStateModel extends StateModel {
+    public BrokerResourceOnlineOfflineStateModel() {
+    }
+
+    @Transition(
+        from = "OFFLINE",
+        to = "ONLINE"
+    )
+    public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
+        LOGGER.info("Become Online from Offline");
+    }
+
+    @Transition(
+        from = "ONLINE",
+        to = "OFFLINE"
+    )
+    public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
+      LOGGER.info("Become Offline from Online");
+
+    }
+
+    @Transition(
+        from = "OFFLINE",
+        to = "DROPPED"
+    )
+    public void onBecomeDroppedFromOffline(Message message, NotificationContext context)
{
+      LOGGER.info("Become Dropped from Offline");
+    }
+
+    @Transition(
+        from = "ONLINE",
+        to = "DROPPED"
+    )
+    public void onBecomeDroppedFromOnline(Message message, NotificationContext context) {
+      LOGGER.info("Become Dropped from Online");
+    }
+
+    @Transition(
+        from = "ERROR",
+        to = "OFFLINE"
+    )
+    public void onBecomeOfflineFromError(Message message, NotificationContext context) {
+      LOGGER.info("Become Offline from Error");
+    }
+  }
+}
+

http://git-wip-us.apache.org/repos/asf/helix/blob/4b50445d/helix-core/src/main/java/org/apache/helix/examples/SegmentOnlineOfflineStateModelFactory.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/examples/SegmentOnlineOfflineStateModelFactory.java
b/helix-core/src/main/java/org/apache/helix/examples/SegmentOnlineOfflineStateModelFactory.java
new file mode 100644
index 0000000..30a23d8
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/examples/SegmentOnlineOfflineStateModelFactory.java
@@ -0,0 +1,53 @@
+package org.apache.helix.examples;
+
+import org.apache.helix.NotificationContext;
+import org.apache.helix.model.Message;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.apache.helix.participant.statemachine.StateModelInfo;
+import org.apache.helix.participant.statemachine.Transition;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class SegmentOnlineOfflineStateModelFactory extends StateModelFactory<StateModel>
{
+  public SegmentOnlineOfflineStateModelFactory() {
+  }
+
+  @Override
+  public StateModel createNewStateModel(String partitionName) {
+    final SegmentOnlineOfflineStateModel SegmentOnlineOfflineStateModel = new SegmentOnlineOfflineStateModel();
+    return SegmentOnlineOfflineStateModel;
+  }
+
+  @StateModelInfo(states = "{'OFFLINE','ONLINE', 'DROPPED'}", initialState = "OFFLINE")
+  public static class SegmentOnlineOfflineStateModel extends StateModel {
+    private static final Logger LOGGER = LoggerFactory.getLogger(SegmentOnlineOfflineStateModel.class);
+
+    @Transition(from = "OFFLINE", to = "ONLINE")
+    public void onBecomeOnlineFromOffline(Message message, NotificationContext context) {
+      LOGGER.info("Become Online from Offline");
+    }
+
+    // Remove segment from InstanceDataManager.
+    // Still keep the data files in local.
+    @Transition(from = "ONLINE", to = "OFFLINE")
+    public void onBecomeOfflineFromOnline(Message message, NotificationContext context) {
+      LOGGER.info("Become Offline from Online");
+
+    }
+
+    // Delete segment from local directory.
+    @Transition(from = "OFFLINE", to = "DROPPED")
+    public void onBecomeDroppedFromOffline(Message message, NotificationContext context)
{
+      LOGGER.info("Become Dropped from Offline");
+
+    }
+
+    @Transition(from = "ONLINE", to = "DROPPED")
+    public void onBecomeDroppedFromOnline(Message message, NotificationContext context) {
+      LOGGER.info("Become Dropped from Online");
+
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/helix/blob/4b50445d/helix-core/src/main/java/org/apache/helix/tools/commandtools/ExampleParticipant.java
----------------------------------------------------------------------
diff --git a/helix-core/src/main/java/org/apache/helix/tools/commandtools/ExampleParticipant.java
b/helix-core/src/main/java/org/apache/helix/tools/commandtools/ExampleParticipant.java
new file mode 100644
index 0000000..3085075
--- /dev/null
+++ b/helix-core/src/main/java/org/apache/helix/tools/commandtools/ExampleParticipant.java
@@ -0,0 +1,207 @@
+package org.apache.helix.tools.commandtools;
+
+/*
+ * 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.
+ */
+
+import java.util.Arrays;
+import java.util.List;
+import org.apache.commons.cli.CommandLine;
+import org.apache.commons.cli.CommandLineParser;
+import org.apache.commons.cli.GnuParser;
+import org.apache.commons.cli.HelpFormatter;
+import org.apache.commons.cli.Option;
+import org.apache.commons.cli.OptionBuilder;
+import org.apache.commons.cli.OptionGroup;
+import org.apache.commons.cli.Options;
+import org.apache.commons.cli.ParseException;
+import org.apache.helix.HelixManager;
+import org.apache.helix.HelixManagerFactory;
+import org.apache.helix.InstanceType;
+import org.apache.helix.examples.BrokerResourceOnlineOfflineStateModelFactory;
+import org.apache.helix.examples.LeaderStandbyStateModelFactory;
+import org.apache.helix.examples.MasterSlaveStateModelFactory;
+import org.apache.helix.examples.OnlineOfflineStateModelFactory;
+import org.apache.helix.examples.SegmentOnlineOfflineStateModelFactory;
+import org.apache.helix.manager.zk.HelixManagerShutdownHook;
+import org.apache.helix.model.Message.MessageType;
+import org.apache.helix.participant.StateMachineEngine;
+import org.apache.helix.participant.statemachine.StateModel;
+import org.apache.helix.participant.statemachine.StateModelFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ExampleParticipant {
+  private static final Logger LOG = LoggerFactory.getLogger(ExampleParticipant.class);
+
+  public static final String zkServer = "zkSvr";
+  public static final String cluster = "cluster";
+  public static final String instances = "instances";
+  public static final String help = "help";
+  public static final String transDelay = "transDelay";
+
+  private final String zkConnectString;
+  private final String clusterName;
+  private final String instanceName;
+  private HelixManager manager;
+
+  private StateModelFactory<StateModel> stateModelFactory;
+  private final int delay;
+
+  public ExampleParticipant(String zkConnectString, String clusterName, String instanceName,
+      int delay) {
+    this.zkConnectString = zkConnectString;
+    this.clusterName = clusterName;
+    this.instanceName = instanceName;
+    this.delay = delay;
+  }
+
+  public void start() throws Exception {
+    manager =
+        HelixManagerFactory.getZKHelixManager(clusterName, instanceName, InstanceType.PARTICIPANT,
+            zkConnectString);
+
+    // genericStateMachineHandler = new StateMachineEngine();
+    // genericStateMachineHandler.registerStateModelFactory(stateModelType,
+    // stateModelFactory);
+
+    StateMachineEngine stateMach = manager.getStateMachineEngine();
+    stateMach.registerStateModelFactory("MasterSlave",
+        new MasterSlaveStateModelFactory(this.instanceName, delay));
+    stateMach.registerStateModelFactory("OnlineOffline",
+        new OnlineOfflineStateModelFactory(this.instanceName, delay));
+    stateMach.registerStateModelFactory("LeaderStandby",
+        new LeaderStandbyStateModelFactory(this.instanceName, delay));
+    stateMach.registerStateModelFactory("BrokerResourceOnlineOfflineStateModel",
+        new BrokerResourceOnlineOfflineStateModelFactory());
+    stateMach.registerStateModelFactory("SegmentOnlineOfflineStateModel",
+        new SegmentOnlineOfflineStateModelFactory());
+
+    manager.connect();
+    manager.getMessagingService()
+        .registerMessageHandlerFactory(MessageType.STATE_TRANSITION.name(), stateMach);
+  }
+
+  public void stop() {
+    manager.disconnect();
+  }
+
+  public HelixManager getManager() {
+    return manager;
+  }
+
+  @SuppressWarnings("static-access")
+  private static Options constructCommandLineOptions() {
+    Option helpOption =
+        OptionBuilder.withLongOpt(help).withDescription("Prints command-line options info")
+            .create();
+
+    Option zkServerOption =
+        OptionBuilder.withLongOpt(zkServer).withDescription("Provide zookeeper address").create();
+    zkServerOption.setArgs(1);
+    zkServerOption.setRequired(true);
+    zkServerOption.setArgName("ZookeeperServerAddress(Required)");
+
+    Option clusterOption =
+        OptionBuilder.withLongOpt(cluster).withDescription("Provide cluster name").create();
+    clusterOption.setArgs(1);
+    clusterOption.setRequired(true);
+    clusterOption.setArgName("Cluster name (Required)");
+
+    Option instancesOption =
+        OptionBuilder.withLongOpt(instances).withDescription("Provide instance names, separated
by ':").create();
+    instancesOption.setArgs(1);
+    instancesOption.setRequired(true);
+    instancesOption.setArgName("Instance names (Required)");
+
+    Option transDelayOption =
+        OptionBuilder.withLongOpt(transDelay).withDescription("Provide state trans delay").create();
+    transDelayOption.setArgs(1);
+    transDelayOption.setRequired(false);
+    transDelayOption.setArgName("Delay time in state transition, in MS");
+
+    OptionGroup optionGroup = new OptionGroup();
+    optionGroup.addOption(zkServerOption);
+
+    Options options = new Options();
+    options.addOption(helpOption);
+    options.addOption(clusterOption);
+    options.addOption(instancesOption);
+    options.addOption(transDelayOption);
+
+    options.addOptionGroup(optionGroup);
+
+    return options;
+  }
+
+  public static void printUsage(Options cliOptions) {
+    HelpFormatter helpFormatter = new HelpFormatter();
+    helpFormatter.setWidth(1000);
+    helpFormatter.printHelp("java " + ExampleParticipant.class.getName(), cliOptions);
+  }
+
+  public static CommandLine processCommandLineArgs(String[] cliArgs) throws Exception {
+    CommandLineParser cliParser = new GnuParser();
+    Options cliOptions = constructCommandLineOptions();
+    try {
+      return cliParser.parse(cliOptions, cliArgs);
+    } catch (ParseException pe) {
+      System.err.println("CommandLineClient: failed to parse command-line options: "
+          + pe.toString());
+      printUsage(cliOptions);
+      System.exit(1);
+    }
+    return null;
+  }
+
+  public static void main(String[] args) throws Exception {
+    int delay = 0;
+
+    CommandLine cmd = processCommandLineArgs(args);
+    String zkConnectString = cmd.getOptionValue(zkServer);
+    String clusterName = cmd.getOptionValue(cluster);
+    String instanceNames = cmd.getOptionValue(instances);
+    List<String> hosts = Arrays.asList(instanceNames.split(":"));
+
+    if (cmd.hasOption(transDelay)) {
+      try {
+        delay = Integer.parseInt(cmd.getOptionValue(transDelay));
+        if (delay < 0) {
+          throw new Exception("delay must be positive");
+        }
+      } catch (Exception e) {
+        e.printStackTrace();
+        delay = 0;
+      }
+    }
+
+    System.out.println("Starting Instances with ZK:" + zkConnectString + ", cluster: " +
clusterName
+        + ", instances: " + hosts);
+
+    for (String instanceName : hosts) {
+      System.out.println("Starting Instance:" + instanceName);
+      ExampleParticipant process =
+          new ExampleParticipant(zkConnectString, clusterName, instanceName, delay);
+      process.start();
+      System.out.println("Started Instance:" + instanceName);
+      Runtime.getRuntime().addShutdownHook(new HelixManagerShutdownHook(process.getManager()));
+    }
+
+    Thread.currentThread().join();
+  }
+}


Mime
View raw message