@Test public void test0PodsClusterReturnsUnknowLeader(VertxTestContext context) { ZookeeperLeaderFinder finder = new ZookeeperLeaderFinder(vertx, null, this::backoff); Checkpoint a = context.checkpoint(); finder.findZookeeperLeader(CLUSTER, NAMESPACE, emptyList(), coKeySecret()) .onComplete(context.succeeding(leader -> { context.verify(() -> assertThat(leader, is(Integer.valueOf(ZookeeperLeaderFinder.UNKNOWN_LEADER)))); a.flag(); })); }
Future<Integer> findZookeeperLeader(String cluster, String namespace, List<Pod> pods, Secret coKeySecret) { if (pods.size() <= 1) { return Future.succeededFuture(pods.size() - 1); } String clusterCaSecretName = KafkaResources.clusterCaCertificateSecretName(cluster); Future<Secret> clusterCaKeySecretFuture = secretOperator.getAsync(namespace, clusterCaSecretName); return clusterCaKeySecretFuture.compose(clusterCaCertificateSecret -> { if (clusterCaCertificateSecret == null) { return Future.failedFuture(Util.missingSecretException(namespace, clusterCaSecretName)); } try { NetClientOptions netClientOptions = clientOptions(coKeySecret, clusterCaCertificateSecret); return zookeeperLeader(cluster, namespace, pods, netClientOptions); } catch (Throwable e) { return Future.failedFuture(e); } }); }
ZookeeperLeaderFinder { Future<Integer> findZookeeperLeader(String cluster, String namespace, List<Pod> pods, Secret coKeySecret) { if (pods.size() <= 1) { return Future.succeededFuture(pods.size() - 1); } String clusterCaSecretName = KafkaResources.clusterCaCertificateSecretName(cluster); Future<Secret> clusterCaKeySecretFuture = secretOperator.getAsync(namespace, clusterCaSecretName); return clusterCaKeySecretFuture.compose(clusterCaCertificateSecret -> { if (clusterCaCertificateSecret == null) { return Future.failedFuture(Util.missingSecretException(namespace, clusterCaSecretName)); } try { NetClientOptions netClientOptions = clientOptions(coKeySecret, clusterCaCertificateSecret); return zookeeperLeader(cluster, namespace, pods, netClientOptions); } catch (Throwable e) { return Future.failedFuture(e); } }); } }
ZookeeperLeaderFinder { Future<Integer> findZookeeperLeader(String cluster, String namespace, List<Pod> pods, Secret coKeySecret) { if (pods.size() <= 1) { return Future.succeededFuture(pods.size() - 1); } String clusterCaSecretName = KafkaResources.clusterCaCertificateSecretName(cluster); Future<Secret> clusterCaKeySecretFuture = secretOperator.getAsync(namespace, clusterCaSecretName); return clusterCaKeySecretFuture.compose(clusterCaCertificateSecret -> { if (clusterCaCertificateSecret == null) { return Future.failedFuture(Util.missingSecretException(namespace, clusterCaSecretName)); } try { NetClientOptions netClientOptions = clientOptions(coKeySecret, clusterCaCertificateSecret); return zookeeperLeader(cluster, namespace, pods, netClientOptions); } catch (Throwable e) { return Future.failedFuture(e); } }); } ZookeeperLeaderFinder(Vertx vertx, SecretOperator secretOperator, Supplier<BackOff> backOffSupplier); }
ZookeeperLeaderFinder { Future<Integer> findZookeeperLeader(String cluster, String namespace, List<Pod> pods, Secret coKeySecret) { if (pods.size() <= 1) { return Future.succeededFuture(pods.size() - 1); } String clusterCaSecretName = KafkaResources.clusterCaCertificateSecretName(cluster); Future<Secret> clusterCaKeySecretFuture = secretOperator.getAsync(namespace, clusterCaSecretName); return clusterCaKeySecretFuture.compose(clusterCaCertificateSecret -> { if (clusterCaCertificateSecret == null) { return Future.failedFuture(Util.missingSecretException(namespace, clusterCaSecretName)); } try { NetClientOptions netClientOptions = clientOptions(coKeySecret, clusterCaCertificateSecret); return zookeeperLeader(cluster, namespace, pods, netClientOptions); } catch (Throwable e) { return Future.failedFuture(e); } }); } ZookeeperLeaderFinder(Vertx vertx, SecretOperator secretOperator, Supplier<BackOff> backOffSupplier);  }
ZookeeperLeaderFinder { Future<Integer> findZookeeperLeader(String cluster, String namespace, List<Pod> pods, Secret coKeySecret) { if (pods.size() <= 1) { return Future.succeededFuture(pods.size() - 1); } String clusterCaSecretName = KafkaResources.clusterCaCertificateSecretName(cluster); Future<Secret> clusterCaKeySecretFuture = secretOperator.getAsync(namespace, clusterCaSecretName); return clusterCaKeySecretFuture.compose(clusterCaCertificateSecret -> { if (clusterCaCertificateSecret == null) { return Future.failedFuture(Util.missingSecretException(namespace, clusterCaSecretName)); } try { NetClientOptions netClientOptions = clientOptions(coKeySecret, clusterCaCertificateSecret); return zookeeperLeader(cluster, namespace, pods, netClientOptions); } catch (Throwable e) { return Future.failedFuture(e); } }); } ZookeeperLeaderFinder(Vertx vertx, SecretOperator secretOperator, Supplier<BackOff> backOffSupplier);  static final int UNKNOWN_LEADER; }