@Test public void testFinderHandlesFailureByLeaderFoundOnThirdAttempt(VertxTestContext context) throws InterruptedException { int desiredLeaderId = 1; int succeedOnAttempt = 2; when(mock.getAsync(eq(NAMESPACE), eq(ClusterOperator.secretName(CLUSTER)))) .thenReturn(Future.succeededFuture( new SecretBuilder() .withNewMetadata() .withName(ClusterOperator.secretName(CLUSTER)) .withNamespace(NAMESPACE) .endMetadata() .withData(map("cluster-operator.key", "notacert", "cluster-operator.crt", "notacert")) .build())); when(mock.getAsync(eq(NAMESPACE), eq(KafkaResources.clusterCaCertificateSecretName(CLUSTER)))) .thenReturn(Future.succeededFuture( new SecretBuilder() .withNewMetadata() .withName(KafkaResources.clusterCaCertificateSecretName(CLUSTER)) .withNamespace(NAMESPACE) .endMetadata() .withData(map(Ca.CA_CRT, "notacert")) .build())); int[] ports = startMockZks(context, 2, (id, attempt) -> attempt == succeedOnAttempt && id == desiredLeaderId); TestingZookeeperLeaderFinder finder = new TestingZookeeperLeaderFinder(this::backoff, ports); Checkpoint a = context.checkpoint(); finder.findZookeeperLeader(CLUSTER, NAMESPACE, asList(createPodWithId(0), createPodWithId(1)), coKeySecret()) .onComplete(context.succeeding(leader -> context.verify(() -> { assertThat(leader, is(desiredLeaderId)); for (FakeZk zk : zks) { assertThat("Unexpected number of attempts for node " + zk.id, zk.attempts.get(), is(succeedOnAttempt + 1)); } 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; }