@Test public void testSecretWithMissingClusterOperatorKeyThrowsException(VertxTestContext context) { SecretOperator mock = mock(SecretOperator.class); ZookeeperLeaderFinder finder = new ZookeeperLeaderFinder(vertx, mock, this::backoff); Mockito.reset(mock); when(mock.getAsync(eq(NAMESPACE), eq(KafkaResources.clusterCaCertificateSecretName(CLUSTER)))) .thenReturn(Future.succeededFuture( new SecretBuilder() .withNewMetadata() .withName(KafkaResources.clusterCaCertificateSecretName(CLUSTER)) .withNamespace(NAMESPACE) .endMetadata() .withData(emptyMap()) .build())); Secret secretWithMissingClusterOperatorKey = new SecretBuilder() .withNewMetadata() .withName(ClusterOperator.secretName(CLUSTER)) .withNamespace(NAMESPACE) .endMetadata() .withData(emptyMap()) .build(); Checkpoint a = context.checkpoint(); finder.findZookeeperLeader(CLUSTER, NAMESPACE, asList(createPodWithId(0), createPodWithId(1)), secretWithMissingClusterOperatorKey) .onComplete(context.failing(e -> context.verify(() -> { assertThat(e, instanceOf(RuntimeException.class)); assertThat(e.getMessage(), is("The Secret testns/testcluster-cluster-operator-certs is missing the key cluster-operator.key")); 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; }