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; }
@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(); })); }
@Test public void test1PodClusterReturnsOnlyPodAsLeader(VertxTestContext context) { ZookeeperLeaderFinder finder = new ZookeeperLeaderFinder(vertx, null, this::backoff); Checkpoint a = context.checkpoint(); int firstPodIndex = 0; finder.findZookeeperLeader(CLUSTER, NAMESPACE, asList(createPodWithId(firstPodIndex)), coKeySecret()) .onComplete(context.succeeding(leader -> { context.verify(() -> assertThat(leader, is(Integer.valueOf(firstPodIndex)))); a.flag(); })); }
@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(); }))); }
@Test public void testSecretsCorrupted(VertxTestContext context) { SecretOperator mock = mock(SecretOperator.class); ZookeeperLeaderFinder finder = new ZookeeperLeaderFinder(vertx, mock, this::backoff); 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())); Secret secretWithBadCertificate = new SecretBuilder() .withNewMetadata() .withName(ClusterOperator.secretName(CLUSTER)) .withNamespace(NAMESPACE) .endMetadata() .withData(map("cluster-operator.key", "notacert", "cluster-operator.crt", "notacert", "cluster-operator.p12", "notatruststore", "cluster-operator.password", "notapassword")) .build(); Checkpoint a = context.checkpoint(); finder.findZookeeperLeader(CLUSTER, NAMESPACE, asList(createPodWithId(0), createPodWithId(1)), secretWithBadCertificate) .onComplete(context.failing(e -> context.verify(() -> { assertThat(e, instanceOf(RuntimeException.class)); assertThat(e.getMessage(), is("Bad/corrupt certificate found in data.cluster-operator\\.crt of Secret testcluster-cluster-operator-certs in namespace testns")); a.flag(); }))); }
@Test public void testReturnUnknownLeaderWhenMaxAttemptsExceeded(VertxTestContext context) throws InterruptedException { String coSecretName = ClusterOperator.secretName(CLUSTER); when(mock.getAsync(eq(NAMESPACE), eq(coSecretName))) .thenReturn(Future.succeededFuture( new SecretBuilder() .withNewMetadata() .withName(coSecretName) .withNamespace(NAMESPACE) .endMetadata() .withData(map("cluster-operator.key", "notacert", "cluster-operator.crt", "notacert")) .build())); String clusterCaSecretName = KafkaResources.clusterCaCertificateSecretName(CLUSTER); when(mock.getAsync(eq(NAMESPACE), eq(clusterCaSecretName))) .thenReturn(Future.succeededFuture( new SecretBuilder() .withNewMetadata() .withName(clusterCaSecretName) .withNamespace(NAMESPACE) .endMetadata() .withData(map(Ca.CA_CRT, "notacert")) .build())); int[] ports = startMockZks(context, 2, (id, attempt) -> false); ZookeeperLeaderFinder 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(ZookeeperLeaderFinder.UNKNOWN_LEADER)); for (FakeZk zk : zks) { assertThat("Unexpected number of attempts for node " + zk.id, zk.attempts.get(), is(MAX_ATTEMPTS + 1)); } a.flag(); }))); }
@Test public void testReturnUnknownLeaderDuringNetworkExceptions(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException { 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) -> false); stopZks(); ZookeeperLeaderFinder 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(ZookeeperLeaderFinder.UNKNOWN_LEADER)); for (FakeZk zk : zks) { assertThat("Unexpected number of attempts for node " + zk.id, zk.attempts.get(), is(0)); } a.flag(); }))); }
@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(); }))); }
@Test public void testLeaderFoundFirstAttempt(VertxTestContext context) throws InterruptedException, ExecutionException, TimeoutException { int leader = 1; when(mock.getAsync(eq(NAMESPACE), eq(ClusterOperator.secretName(CLUSTER)))) .thenAnswer(i -> 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)))) .thenAnswer(i -> 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) -> id == leader); ZookeeperLeaderFinder finder = new TestingZookeeperLeaderFinder(this::backoff, ports); Checkpoint a = context.checkpoint(); finder.findZookeeperLeader(CLUSTER, NAMESPACE, asList(createPodWithId(0), createPodWithId(1)), coKeySecret()) .onComplete(context.succeeding(l -> context.verify(() -> { assertThat(l, is(leader)); for (FakeZk zk : zks) { assertThat("Unexpected number of attempts for node " + zk.id, zk.attempts.get(), is(1)); } a.flag(); }))); }