本文整理了Java中org.apache.samza.zk.ZkLeaderElector类的一些代码示例，展示了ZkLeaderElector类的具体用法。这些代码示例主要来源于Github/Stackoverflow/Maven等平台，是从一些精选项目中提取出来的代码，具有较强的参考意义，能在一定程度帮忙到你。ZkLeaderElector类的具体详情如下：
包路径：org.apache.samza.zk.ZkLeaderElector
类名称：ZkLeaderElector

ZkLeaderElector介绍

[英]An implementation of Leader Elector using Zookeeper. Each participant in the leader election process creates an instance of this class and tries to become the leader. The participant with the lowest sequence number in the ZK subtree for election becomes the leader. Every non-leader sets a watcher on its predecessor, where the predecessor is the participant with the largest sequence number that is less than the current participant's sequence number.
[中]使用Zookeeper实现领导人选举。领导者选举过程中的每个参与者都会创建该类的一个实例，并尝试成为领导者。在ZK子树中，在选举中序列号最低的参与者成为领导者。每个非领导者在其前任上设置一个观察者，其中前任是最大序列号小于当前参与者序列号的参与者。

代码示例

代码示例来源：origin: apache/samza

@Override
public LeaderElector getLeaderElector() {
 return new ZkLeaderElector(processorIdStr, zkUtils);
}

代码示例来源：origin: apache/samza

public void onProcessorChange(List<String> processors) {
 if (leaderElector.amILeader()) {
  LOG.info("ZkJobCoordinator::onProcessorChange - list of processors changed. List size=" + processors.size());
  debounceTimer.scheduleAfterDebounceTime(ON_PROCESSOR_CHANGE, debounceTimeMs, this::doOnProcessorChange);
 }
}

代码示例来源：origin: apache/samza

@Override
 public void doHandleDataDeleted(String dataPath) {
  LOG.info(zLog("Data deleted on path " + dataPath + ". Predecessor went away. So, trying to become leader."));
  tryBecomeLeader();
 }
}

代码示例来源：origin: apache/samza

@Test
public void testUnregisteredProcessorInLeaderElection() {
 String processorId = "1";
 ZkUtils mockZkUtils = mock(ZkUtils.class);
 when(mockZkUtils.getSortedActiveProcessorsZnodes()).thenReturn(new ArrayList<String>());
 Mockito.doNothing().when(mockZkUtils).validatePaths(any(String[].class));
 ZkKeyBuilder kb = mock(ZkKeyBuilder.class);
 when(kb.getProcessorsPath()).thenReturn("");
 when(mockZkUtils.getKeyBuilder()).thenReturn(kb);
 ZkLeaderElector leaderElector = new ZkLeaderElector(processorId, mockZkUtils, null);
 leaderElector.setLeaderElectorListener(() -> { });
 try {
  leaderElector.tryBecomeLeader();
  Assert.fail("Was expecting leader election to fail!");
 } catch (SamzaException e) {
  // No-op Expected
 }
}

代码示例来源：origin: org.apache.samza/samza-core_2.10

if (leaderElector.amILeader()) {
 LOG.info("Resigning leadership for processorId: " + processorId);
 leaderElector.resignLeadership();

代码示例来源：origin: apache/samza

@Test
public void testAmILeader() {
 BooleanResult isLeader1 = new BooleanResult();
 BooleanResult isLeader2 = new BooleanResult();
 // Processor-1
 ZkUtils zkUtils1 = getZkUtilsWithNewClient();
 ZkLeaderElector leaderElector1 = new ZkLeaderElector("1", zkUtils1, null);
 leaderElector1.setLeaderElectorListener(() -> isLeader1.res = true);
 // Processor-2
 ZkUtils zkUtils2 = getZkUtilsWithNewClient();
 ZkLeaderElector leaderElector2 = new ZkLeaderElector("2", zkUtils2, null);
 leaderElector2.setLeaderElectorListener(() -> isLeader2.res = true);
 // Before Leader Election
 Assert.assertFalse(leaderElector1.amILeader());
 Assert.assertFalse(leaderElector2.amILeader());
 leaderElector1.tryBecomeLeader();
 leaderElector2.tryBecomeLeader();
 // After Leader Election
 Assert.assertTrue(leaderElector1.amILeader());
 Assert.assertFalse(leaderElector2.amILeader());
 zkUtils1.close();
 zkUtils2.close();
}

代码示例来源：origin: org.apache.samza/samza-core_2.12

ZkJobCoordinator(Config config, MetricsRegistry metricsRegistry, ZkUtils zkUtils) {
 this.config = config;
 this.metrics = new ZkJobCoordinatorMetrics(metricsRegistry);
 this.processorId = createProcessorId(config);
 this.zkUtils = zkUtils;
 // setup a listener for a session state change
 // we are mostly interested in "session closed" and "new session created" events
 zkUtils.getZkClient().subscribeStateChanges(new ZkSessionStateChangedListener());
 leaderElector = new ZkLeaderElector(processorId, zkUtils);
 leaderElector.setLeaderElectorListener(new LeaderElectorListenerImpl());
 this.debounceTimeMs = new JobConfig(config).getDebounceTimeMs();
 this.reporters = MetricsReporterLoader.getMetricsReporters(new MetricsConfig(config), processorId);
 debounceTimer = new ScheduleAfterDebounceTime(processorId);
 debounceTimer.setScheduledTaskCallback(throwable -> {
   LOG.error("Received exception in debounce timer! Stopping the job coordinator", throwable);
   stop();
  });
 this.barrier =  new ZkBarrierForVersionUpgrade(zkUtils.getKeyBuilder().getJobModelVersionBarrierPrefix(), zkUtils, new ZkBarrierListenerImpl(), debounceTimer);
 systemAdmins = new SystemAdmins(config);
 streamMetadataCache = new StreamMetadataCache(systemAdmins, METADATA_CACHE_TTL_MS, SystemClock.instance());
}

代码示例来源：origin: apache/samza

LOG.debug(zLog("Current active processors - " + children));
int index = children.indexOf(ZkKeyBuilder.parseIdFromPath(currentPath));
 LOG.info(zLog("Eligible to become the leader!"));
 if (leaderElectorListener != null) {
  leaderElectorListener.onBecomingLeader();
if (!predecessor.equals(currentSubscription)) {
 if (currentSubscription != null) {
  LOG.debug(zLog("Unsubscribing data change for " + currentSubscription));
  zkUtils.unsubscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription,
    previousProcessorChangeListener);
 LOG.info(zLog("Subscribing data change for " + predecessor));
 zkUtils.subscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription,
   previousProcessorChangeListener);
 LOG.info(zLog("Predecessor still exists. Current subscription is valid. Continuing as non-leader."));
} else {
 try {
  Thread.interrupted();
 LOG.info(zLog("Predecessor doesn't exist anymore. Trying to become leader again..."));
 tryBecomeLeader();

代码示例来源：origin: apache/samza

@VisibleForTesting
public ZkLeaderElector(String processorIdStr,
            ZkUtils zkUtils,
            IZkDataListener previousProcessorChangeListener) {
 this.processorIdStr = processorIdStr;
 this.zkUtils = zkUtils;
 this.keyBuilder = zkUtils.getKeyBuilder();
 this.hostName = getHostName();
 this.previousProcessorChangeListener = previousProcessorChangeListener;
}

代码示例来源：origin: apache/samza

@Test
public void testLeaderElectionRegistersProcessor() {
 List<String> activeProcessors = new ArrayList<String>() {
  {
   add("0000000000");
  }
 };
 ZkUtils mockZkUtils = mock(ZkUtils.class);
 when(mockZkUtils.registerProcessorAndGetId(any())).
   thenReturn(KEY_BUILDER.getProcessorsPath() + "/0000000000");
 when(mockZkUtils.getSortedActiveProcessorsZnodes()).thenReturn(activeProcessors);
 Mockito.doNothing().when(mockZkUtils).validatePaths(any(String[].class));
 ZkKeyBuilder kb = mock(ZkKeyBuilder.class);
 when(kb.getProcessorsPath()).thenReturn("");
 when(mockZkUtils.getKeyBuilder()).thenReturn(kb);
 ZkLeaderElector leaderElector = new ZkLeaderElector("1", mockZkUtils, null);
 BooleanResult isLeader = new BooleanResult();
 leaderElector.setLeaderElectorListener(() -> isLeader.res = true);
 leaderElector.tryBecomeLeader();
 Assert.assertTrue(TestZkUtils.testWithDelayBackOff(() -> isLeader.res, 2, 100));
}

代码示例来源：origin: org.apache.samza/samza-core_2.11

if (leaderElector.amILeader()) {
 LOG.info("Resigning leadership for processorId: " + processorId);
 leaderElector.resignLeadership();

代码示例来源：origin: apache/samza

ZkLeaderElector leaderElector1 = new ZkLeaderElector("processor1", zkUtils1, new IZkDataListener() {
 @Override
 public void handleDataChange(String dataPath, Object data)
leaderElector1.setLeaderElectorListener(() -> isLeader1.res = true);
ZkLeaderElector leaderElector2 = new ZkLeaderElector("processor2", zkUtils2, new IZkDataListener() {
 @Override
 public void handleDataChange(String dataPath, Object data)
leaderElector2.setLeaderElectorListener(() -> isLeader2.res = true);
ZkLeaderElector leaderElector3 = new ZkLeaderElector("processor3", zkUtils3, new IZkDataListener() {
 @Override
 public void handleDataChange(String dataPath, Object data)
leaderElector3.setLeaderElectorListener(() -> isLeader3.res = true);
leaderElector1.tryBecomeLeader();
leaderElector2.tryBecomeLeader();
leaderElector3.tryBecomeLeader();
Assert.assertFalse(TestZkUtils.testWithDelayBackOff(() -> isLeader3.res, 2, 100));
Assert.assertTrue(leaderElector1.amILeader());
Assert.assertFalse(leaderElector2.amILeader());
Assert.assertFalse(leaderElector3.amILeader());

代码示例来源：origin: org.apache.samza/samza-core_2.10

ZkJobCoordinator(Config config, MetricsRegistry metricsRegistry, ZkUtils zkUtils) {
 this.config = config;
 this.metrics = new ZkJobCoordinatorMetrics(metricsRegistry);
 this.processorId = createProcessorId(config);
 this.zkUtils = zkUtils;
 // setup a listener for a session state change
 // we are mostly interested in "session closed" and "new session created" events
 zkUtils.getZkClient().subscribeStateChanges(new ZkSessionStateChangedListener());
 leaderElector = new ZkLeaderElector(processorId, zkUtils);
 leaderElector.setLeaderElectorListener(new LeaderElectorListenerImpl());
 this.debounceTimeMs = new JobConfig(config).getDebounceTimeMs();
 this.reporters = MetricsReporterLoader.getMetricsReporters(new MetricsConfig(config), processorId);
 debounceTimer = new ScheduleAfterDebounceTime(processorId);
 debounceTimer.setScheduledTaskCallback(throwable -> {
   LOG.error("Received exception in debounce timer! Stopping the job coordinator", throwable);
   stop();
  });
 this.barrier =  new ZkBarrierForVersionUpgrade(zkUtils.getKeyBuilder().getJobModelVersionBarrierPrefix(), zkUtils, new ZkBarrierListenerImpl(), debounceTimer);
 systemAdmins = new SystemAdmins(config);
 streamMetadataCache = new StreamMetadataCache(systemAdmins, METADATA_CACHE_TTL_MS, SystemClock.instance());
}

代码示例来源：origin: org.apache.samza/samza-core

LOG.debug(zLog("Current active processors - " + children));
int index = children.indexOf(ZkKeyBuilder.parseIdFromPath(currentPath));
 LOG.info(zLog("Eligible to become the leader!"));
 if (leaderElectorListener != null) {
  leaderElectorListener.onBecomingLeader();
if (!predecessor.equals(currentSubscription)) {
 if (currentSubscription != null) {
  LOG.debug(zLog("Unsubscribing data change for " + currentSubscription));
  zkUtils.unsubscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription,
    previousProcessorChangeListener);
 LOG.info(zLog("Subscribing data change for " + predecessor));
 zkUtils.subscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription,
   previousProcessorChangeListener);
 LOG.info(zLog("Predecessor still exists. Current subscription is valid. Continuing as non-leader."));
} else {
 try {
  Thread.interrupted();
 LOG.info(zLog("Predecessor doesn't exist anymore. Trying to become leader again..."));
 tryBecomeLeader();

代码示例来源：origin: org.apache.samza/samza-core_2.12

@VisibleForTesting
public ZkLeaderElector(String processorIdStr,
            ZkUtils zkUtils,
            IZkDataListener previousProcessorChangeListener) {
 this.processorIdStr = processorIdStr;
 this.zkUtils = zkUtils;
 this.keyBuilder = zkUtils.getKeyBuilder();
 this.hostName = getHostName();
 this.previousProcessorChangeListener = previousProcessorChangeListener;
}

代码示例来源：origin: apache/samza

ZkLeaderElector leaderElector1 = new ZkLeaderElector("1", zkUtils1, null);
leaderElector1.setLeaderElectorListener(() -> isLeader1.res = true);
ZkLeaderElector leaderElector2 = new ZkLeaderElector("2", zkUtils2, null);
leaderElector2.setLeaderElectorListener(() -> isLeader2.res = true);
ZkLeaderElector leaderElector3 = new ZkLeaderElector("3", zkUtils3, null);
leaderElector3.setLeaderElectorListener(() -> isLeader3.res = true);
leaderElector1.tryBecomeLeader();
leaderElector2.tryBecomeLeader();
leaderElector3.tryBecomeLeader();

代码示例来源：origin: org.apache.samza/samza-core

if (leaderElector.amILeader()) {
 LOG.info("Resigning leadership for processorId: " + processorId);
 leaderElector.resignLeadership();

代码示例来源：origin: org.apache.samza/samza-core_2.11

ZkJobCoordinator(Config config, MetricsRegistry metricsRegistry, ZkUtils zkUtils) {
 this.config = config;
 this.metrics = new ZkJobCoordinatorMetrics(metricsRegistry);
 this.processorId = createProcessorId(config);
 this.zkUtils = zkUtils;
 // setup a listener for a session state change
 // we are mostly interested in "session closed" and "new session created" events
 zkUtils.getZkClient().subscribeStateChanges(new ZkSessionStateChangedListener());
 leaderElector = new ZkLeaderElector(processorId, zkUtils);
 leaderElector.setLeaderElectorListener(new LeaderElectorListenerImpl());
 this.debounceTimeMs = new JobConfig(config).getDebounceTimeMs();
 this.reporters = MetricsReporterLoader.getMetricsReporters(new MetricsConfig(config), processorId);
 debounceTimer = new ScheduleAfterDebounceTime(processorId);
 debounceTimer.setScheduledTaskCallback(throwable -> {
   LOG.error("Received exception in debounce timer! Stopping the job coordinator", throwable);
   stop();
  });
 this.barrier =  new ZkBarrierForVersionUpgrade(zkUtils.getKeyBuilder().getJobModelVersionBarrierPrefix(), zkUtils, new ZkBarrierListenerImpl(), debounceTimer);
 systemAdmins = new SystemAdmins(config);
 streamMetadataCache = new StreamMetadataCache(systemAdmins, METADATA_CACHE_TTL_MS, SystemClock.instance());
}

代码示例来源：origin: org.apache.samza/samza-core_2.10

LOG.debug(zLog("Current active processors - " + children));
int index = children.indexOf(ZkKeyBuilder.parseIdFromPath(currentPath));
 LOG.info(zLog("Eligible to become the leader!"));
 if (leaderElectorListener != null) {
  leaderElectorListener.onBecomingLeader();
if (!predecessor.equals(currentSubscription)) {
 if (currentSubscription != null) {
  LOG.debug(zLog("Unsubscribing data change for " + currentSubscription));
  zkUtils.unsubscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription,
    previousProcessorChangeListener);
 LOG.info(zLog("Subscribing data change for " + predecessor));
 zkUtils.subscribeDataChanges(keyBuilder.getProcessorsPath() + "/" + currentSubscription,
   previousProcessorChangeListener);
 LOG.info(zLog("Predecessor still exists. Current subscription is valid. Continuing as non-leader."));
} else {
 try {
  Thread.interrupted();
 LOG.info(zLog("Predecessor doesn't exist anymore. Trying to become leader again..."));
 tryBecomeLeader();

代码示例来源：origin: org.apache.samza/samza-core_2.10

public void onProcessorChange(List<String> processors) {
 if (leaderElector.amILeader()) {
  LOG.info("ZkJobCoordinator::onProcessorChange - list of processors changed. List size=" + processors.size());
  debounceTimer.scheduleAfterDebounceTime(ON_PROCESSOR_CHANGE, debounceTimeMs, () -> doOnProcessorChange(processors));
 }
}