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

ZKSegmentContainerMonitor介绍

[英]Monitors the current set of running segment containers and ensure it matches the ownership assignment for this host. This monitor watches the shared zk entry that contains the segment container ownership information and starts or stops appropriate segment containers locally. Any start failures are periodically retried until the desired ownership state is achieved.
[中]监视当前正在运行的段容器集，并确保它与此主机的所有权分配相匹配。此监视器监视包含段容器所有权信息的共享zk条目，并在本地启动或停止相应的段容器。任何启动失败都会定期重试，直到达到所需的所有权状态。

代码示例

代码示例来源：origin: pravega/pravega

/**
 * Creates a new instance of the ZKSegmentContainerManager class.
 *
 * @param containerRegistry      The SegmentContainerRegistry to manage.
 * @param zkClient               ZooKeeper client.
 * @param pravegaServiceEndpoint Pravega service endpoint details.
 * @param executor               Executor service for running async operations.
 */
ZKSegmentContainerManager(SegmentContainerRegistry containerRegistry, CuratorFramework zkClient,
             Host pravegaServiceEndpoint, ScheduledExecutorService executor) {
  Preconditions.checkNotNull(containerRegistry, "containerRegistry");
  Preconditions.checkNotNull(zkClient, "zkClient");
  this.host = Preconditions.checkNotNull(pravegaServiceEndpoint, "pravegaServiceEndpoint");
  this.cluster = new ClusterZKImpl(zkClient, ClusterType.HOST);
  this.containerMonitor = new ZKSegmentContainerMonitor(containerRegistry, zkClient, pravegaServiceEndpoint, executor);
}

代码示例来源：origin: pravega/pravega

/**
 * Initialize the monitor. This will start the monitor thread which will process the start/stop container events.
 */
public void initialize() {
  initialize(MONITOR_INTERVAL);
}

代码示例来源：origin: pravega/pravega

unregisterHandle(containerId);
} finally {

代码示例来源：origin: pravega/pravega

/**
 * Test if no mapping is present in zk.
 *
 * @throws Exception if an error occurred.
 */
@Test
public void testInitializeNoMapping() throws Exception {
  @Cleanup
  CuratorFramework zkClient = startClient();
  @Cleanup
  ZKSegmentContainerMonitor segMonitor = createContainerMonitor(createMockContainerRegistry(), zkClient);
  segMonitor.initialize();
  assertEquals("Unexpected number of handles.", 0, segMonitor.getRegisteredContainers().size());
}

代码示例来源：origin: pravega/pravega

@Test
public void testClose() throws Exception {
  @Cleanup
  CuratorFramework zkClient = startClient();
  initializeHostContainerMapping(zkClient);
  SegmentContainerRegistry containerRegistry = mock(SegmentContainerRegistry.class);
  ContainerHandle containerHandle1 = mock(ContainerHandle.class);
  when(containerHandle1.getContainerId()).thenReturn(1);
  when(containerRegistry.startContainer(eq(1), any()))
      .thenReturn(CompletableFuture.completedFuture(containerHandle1));
  when(containerRegistry.stopContainer(any(), any())).thenReturn(CompletableFuture.completedFuture(null));
  ZKSegmentContainerMonitor segMonitor = createContainerMonitor(containerRegistry, zkClient);
  segMonitor.initialize(Duration.ofSeconds(1));
  segMonitor.close();
  assertEquals(0, segMonitor.getRegisteredContainers().size());
}

代码示例来源：origin: pravega/pravega

Set<Integer> desiredList = getDesiredContainerList();
if (desiredList != null) {
  Collection<Integer> runningContainers = new HashSet<>(this.handles.keySet());

代码示例来源：origin: pravega/pravega

@Cleanup
ZKSegmentContainerMonitor segMonitor = createContainerMonitor(containerRegistry, zkClient);
segMonitor.initialize(Duration.ofSeconds(1));
assertEquals(0, segMonitor.getRegisteredContainers().size());
assertEquals(1, segMonitor.getRegisteredContainers().size());

代码示例来源：origin: pravega/pravega

@Cleanup
ZKSegmentContainerMonitor segMonitor = createContainerMonitor(containerRegistry, zkClient);
segMonitor.initialize(Duration.ofSeconds(1));
assertEquals(0, segMonitor.getRegisteredContainers().size());
assertEquals(0, segMonitor.getRegisteredContainers().size());

代码示例来源：origin: pravega/pravega

@Override
public void initialize() {
  Exceptions.checkNotClosed(closed.get(), this);
  long traceId = LoggerHelpers.traceEnter(log, "initialize");
  try {
    // Initialize the container monitor.
    this.containerMonitor.initialize();
    // Advertise this segment store to the cluster.
    this.cluster.registerHost(this.host);
    log.info("Initialized.");
    LoggerHelpers.traceLeave(log, "initialize", traceId);
  } catch (Exception ex) {
    // Need to make sure we clean up resources if we failed to initialize.
    log.error("Initialization error. Cleaning up.", ex);
    close();
    throw ex;
  }
}

代码示例来源：origin: pravega/pravega

private ZKSegmentContainerMonitor createContainerMonitor(
    SegmentContainerRegistry registry, CuratorFramework zkClient) {
  return new ZKSegmentContainerMonitor(registry, zkClient, PRAVEGA_SERVICE_ENDPOINT, executorService());
}

代码示例来源：origin: pravega/pravega

@Override
public void close() {
  Preconditions.checkState(closed.compareAndSet(false, true));
  try {
    this.hostContainerMapNode.close();
  } catch (IOException e) {
    // Ignoring exception on shutdown.
    log.warn("Failed to close hostContainerMapNode {}", e);
  }
  val task = this.assigmentTask.getAndSet(null);
  if (task != null) {
    task.cancel(true);
  }
  ArrayList<ContainerHandle> toClose = new ArrayList<>(this.handles.values());
  ArrayList<CompletableFuture<Void>> results = new ArrayList<>();
  for (ContainerHandle handle : toClose) {
    results.add(this.registry.stopContainer(handle, CLOSE_TIMEOUT_PER_CONTAINER)
                 .thenAccept(v -> unregisterHandle(handle.getContainerId())));
  }
  // Wait for all the containers to be closed.
  Futures.await(Futures.allOf(results), CLOSE_TIMEOUT_PER_CONTAINER.toMillis());
}

代码示例来源：origin: pravega/pravega

@Cleanup
ZKSegmentContainerMonitor segMonitor = createContainerMonitor(containerRegistry, zkClient);
segMonitor.initialize(Duration.ofSeconds(1));
assertEquals(1, segMonitor.getRegisteredContainers().size());
assertTrue(segMonitor.getRegisteredContainers().contains(2));
assertEquals(1, segMonitor.getRegisteredContainers().size());
assertTrue(segMonitor.getRegisteredContainers().contains(1));

代码示例来源：origin: pravega/pravega

/**
 * Tests if we cannot connect to ZooKeeper (the exception must be propagated to the caller).
 *
 * @throws Exception if an error occurred.
 */
@Test
public void testInitializeError() throws Exception {
  @Cleanup
  CuratorFramework zkClient = startClient();
  @Cleanup
  ZKSegmentContainerMonitor segMonitor = createContainerMonitor(createMockContainerRegistry(), zkClient);
  zkClient.close();
  AssertExtensions.assertThrows(
      "initialize() did not throw an exception when ZooKeeper could not be accessed.",
      () -> segMonitor.initialize(),
      ex -> true); // Any exception will do, as long as it is propagated.
}

代码示例来源：origin: pravega/pravega

@Test
public void testRetryOnStartFailures() throws Exception {
  @Cleanup
  CuratorFramework zkClient = startClient();
  initializeHostContainerMapping(zkClient);
  SegmentContainerRegistry containerRegistry = createMockContainerRegistry();
  @Cleanup
  ZKSegmentContainerMonitor segMonitor = createContainerMonitor(containerRegistry, zkClient);
  segMonitor.initialize(Duration.ofSeconds(1));
  // Simulate a container that fails to start.
  CompletableFuture<ContainerHandle> failedFuture = Futures.failedFuture(new RuntimeException());
  when(containerRegistry.startContainer(eq(2), any()))
      .thenReturn(failedFuture);
  // Use ZK to send that information to the Container Manager.
  HashMap<Host, Set<Integer>> currentData = deserialize(zkClient, PATH);
  currentData.put(PRAVEGA_SERVICE_ENDPOINT, Collections.singleton(2));
  zkClient.setData().forPath(PATH, SerializationUtils.serialize(currentData));
  // Verify that it does not start.
  verify(containerRegistry, timeout(1000).atLeastOnce()).startContainer(eq(2), any());
  assertEquals(0, segMonitor.getRegisteredContainers().size());
  // Now simulate success for the same container.
  ContainerHandle containerHandle = mock(ContainerHandle.class);
  when(containerHandle.getContainerId()).thenReturn(2);
  when(containerRegistry.startContainer(eq(2), any()))
      .thenReturn(CompletableFuture.completedFuture(containerHandle));
  // Verify that it retries and starts the same container again.
  verify(containerRegistry, timeout(1000).atLeastOnce()).startContainer(eq(2), any());
  Thread.sleep(2000);
  assertEquals(1, segMonitor.getRegisteredContainers().size());
}