Skip to content

Commit 08e2993

Browse files
franvilafranvila
authored
added Tls system test (kroxylicious#2104)
* added Tls system test Signed-off-by: Francisco Vila <fvila@redhat.com> * remove unused dependencies Signed-off-by: Francisco Vila <fvila@redhat.com> * fix format Signed-off-by: Francisco Vila <fvila@redhat.com> * refactored certificate gotten from listeners Signed-off-by: Francisco Vila <fvila@redhat.com> * revert kubernetes upgrade Signed-off-by: Francisco Vila <fvila@redhat.com> * rename deployment to CR and added getBootstrap for Kafka CR Signed-off-by: Francisco Vila <fvila@redhat.com> * rename java doc Signed-off-by: Francisco Vila <fvila@redhat.com> * fix format Signed-off-by: Francisco Vila <fvila@redhat.com> * fix trust anchor ref Signed-off-by: Francisco Vila <fvila@redhat.com> * set default value for kafka bootstrap Signed-off-by: Francisco Vila <fvila@redhat.com> * set default value for kafka bootstratp Signed-off-by: Francisco Vila <fvila@redhat.com> * set default value for kafka bootstratp Signed-off-by: Francisco Vila <fvila@redhat.com> * stabilize operator change detection ST Signed-off-by: Francisco Vila <fvila@redhat.com> * revert ST change Signed-off-by: Francisco Vila <fvila@redhat.com> --------- Signed-off-by: Francisco Vila <fvila@redhat.com>
1 parent 989c08d commit 08e2993

File tree

9 files changed

+189
-36
lines changed

9 files changed

+189
-36
lines changed

‎kroxylicious-systemtests/src/main/java/io/kroxylicious/systemtests/Constants.java‎

Lines changed: 3 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,8 @@ private Constants() {
2525
public static final String KROXYLICIOUS_PROXY_SIMPLE_NAME = "simple";
2626
public static final String KROXYLICIOUS_INGRESS_CLUSTER_IP = "cluster-ip";
2727
public static final String KROXYLICIOUS_ENCRYPTION_FILTER_NAME = "encryption";
28+
public static final String KROXYLICIOUS_TLS_CLIENT_CA_CERT = "my-cluster-clients-ca-cert";
29+
public static final String KROXYLICIOUS_TLS_CA_NAME = "ca.pem";
2830

2931
/**
3032
* Strimzi cluster operator deployment name
@@ -102,7 +104,6 @@ private Constants() {
102104
public static final String SERVICE = "Service";
103105
public static final String SERVICE_ACCOUNT = "ServiceAccount";
104106
public static final String STRIMZI_KAFKA_KIND = "Kafka";
105-
public static final String STRIMZI_KAFKA_USER_KIND = "KafkaUser";
106107
public static final String STRIMZI_KAFKA_NODE_POOL_KIND = "KafkaNodePool";
107108

108109
/**
@@ -117,16 +118,10 @@ private Constants() {
117118
*/
118119
public static final String CERT_MANAGER_URL = "https://github.com/cert-manager/cert-manager/releases/latest/download/cert-manager.yaml";
119120
/**
120-
* kafka consumer client label to identify the consumer test client
121+
* the kubernetes labels used to identify the test kafka clients pods
121122
*/
122123
public static final String KAFKA_CONSUMER_CLIENT_LABEL = "kafka-consumer-client";
123-
/**
124-
* kafka producer client label to identify the producer test client
125-
*/
126124
public static final String KAFKA_PRODUCER_CLIENT_LABEL = "kafka-producer-client";
127-
/**
128-
* kafka admin client label to identify the admin test client
129-
*/
130125
public static final String KAFKA_ADMIN_CLIENT_LABEL = "admin-client-cli";
131126
/**
132127
* Image pull policies

‎kroxylicious-systemtests/src/main/java/io/kroxylicious/systemtests/installation/kroxylicious/Kroxylicious.java‎

Lines changed: 40 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -13,18 +13,22 @@
1313
import org.slf4j.Logger;
1414
import org.slf4j.LoggerFactory;
1515

16+
import io.strimzi.api.kafka.model.kafka.listener.ListenerStatus;
17+
1618
import io.kroxylicious.kms.service.TestKmsFacade;
1719
import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaCluster;
1820
import io.kroxylicious.kubernetes.api.v1alpha1.VirtualKafkaClusterStatus;
1921
import io.kroxylicious.kubernetes.api.v1alpha1.virtualkafkaclusterstatus.Ingresses;
2022
import io.kroxylicious.systemtests.Constants;
2123
import io.kroxylicious.systemtests.resources.kms.ExperimentalKmsConfig;
2224
import io.kroxylicious.systemtests.resources.manager.ResourceManager;
25+
import io.kroxylicious.systemtests.templates.kroxylicious.KroxyliciousConfigMapTemplates;
2326
import io.kroxylicious.systemtests.templates.kroxylicious.KroxyliciousFilterTemplates;
2427
import io.kroxylicious.systemtests.templates.kroxylicious.KroxyliciousKafkaClusterRefTemplates;
2528
import io.kroxylicious.systemtests.templates.kroxylicious.KroxyliciousKafkaProxyIngressTemplates;
2629
import io.kroxylicious.systemtests.templates.kroxylicious.KroxyliciousKafkaProxyTemplates;
2730
import io.kroxylicious.systemtests.templates.kroxylicious.KroxyliciousVirtualKafkaClusterTemplates;
31+
import io.kroxylicious.systemtests.utils.KafkaUtils;
2832

2933
import static io.kroxylicious.systemtests.k8s.KubeClusterResource.kubeClient;
3034
import static org.awaitility.Awaitility.await;
@@ -60,11 +64,11 @@ private void createRecordEncryptionFilterConfigMap(TestKmsFacade<?, ?, ?> testKm
6064
*/
6165
public void deployPortIdentifiesNodeWithFilters(String clusterName, List<String> filterNames) {
6266
resourceManager.createResourceFromBuilderWithWait(
63-
KroxyliciousKafkaProxyTemplates.defaultKafkaProxyDeployment(deploymentNamespace, Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME),
64-
KroxyliciousKafkaProxyIngressTemplates.defaultKafkaProxyIngressDeployment(deploymentNamespace, Constants.KROXYLICIOUS_INGRESS_CLUSTER_IP,
67+
KroxyliciousKafkaProxyTemplates.defaultKafkaProxyCR(deploymentNamespace, Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME),
68+
KroxyliciousKafkaProxyIngressTemplates.defaultKafkaProxyIngressCR(deploymentNamespace, Constants.KROXYLICIOUS_INGRESS_CLUSTER_IP,
6569
Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME),
66-
KroxyliciousKafkaClusterRefTemplates.defaultKafkaClusterRefDeployment(deploymentNamespace, clusterName),
67-
KroxyliciousVirtualKafkaClusterTemplates.virtualKafkaClusterWithFilterDeployment(deploymentNamespace, clusterName,
70+
KroxyliciousKafkaClusterRefTemplates.defaultKafkaClusterRefCR(deploymentNamespace, clusterName),
71+
KroxyliciousVirtualKafkaClusterTemplates.virtualKafkaClusterWithFilterCR(deploymentNamespace, clusterName,
6872
Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME,
6973
clusterName, Constants.KROXYLICIOUS_INGRESS_CLUSTER_IP, filterNames));
7074
}
@@ -74,14 +78,42 @@ public void deployPortIdentifiesNodeWithFilters(String clusterName, List<String>
7478
*/
7579
public void deployPortIdentifiesNodeWithNoFilters(String clusterName) {
7680
resourceManager.createResourceFromBuilder(
77-
KroxyliciousKafkaProxyTemplates.defaultKafkaProxyDeployment(deploymentNamespace, Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME),
78-
KroxyliciousKafkaProxyIngressTemplates.defaultKafkaProxyIngressDeployment(deploymentNamespace, Constants.KROXYLICIOUS_INGRESS_CLUSTER_IP,
81+
KroxyliciousKafkaProxyTemplates.defaultKafkaProxyCR(deploymentNamespace, Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME),
82+
KroxyliciousKafkaProxyIngressTemplates.defaultKafkaProxyIngressCR(deploymentNamespace, Constants.KROXYLICIOUS_INGRESS_CLUSTER_IP,
83+
Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME),
84+
KroxyliciousKafkaClusterRefTemplates.defaultKafkaClusterRefCR(deploymentNamespace, clusterName),
85+
KroxyliciousVirtualKafkaClusterTemplates.defaultVirtualKafkaClusterCR(deploymentNamespace, clusterName, Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME,
86+
clusterName, Constants.KROXYLICIOUS_INGRESS_CLUSTER_IP));
87+
}
88+
89+
/**
90+
* Deploy port identifies node with tls and no filters.
91+
*
92+
* @param clusterName the cluster name
93+
*/
94+
public void deployPortIdentifiesNodeWithTlsAndNoFilters(String clusterName) {
95+
createCertificateConfigMap(deploymentNamespace);
96+
resourceManager.createResourceFromBuilder(
97+
KroxyliciousKafkaProxyTemplates.defaultKafkaProxyCR(deploymentNamespace, Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME),
98+
KroxyliciousKafkaProxyIngressTemplates.defaultKafkaProxyIngressCR(deploymentNamespace, Constants.KROXYLICIOUS_INGRESS_CLUSTER_IP,
7999
Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME),
80-
KroxyliciousKafkaClusterRefTemplates.defaultKafkaClusterRefDeployment(deploymentNamespace, clusterName),
81-
KroxyliciousVirtualKafkaClusterTemplates.defaultVirtualKafkaClusterDeployment(deploymentNamespace, clusterName, Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME,
100+
KroxyliciousKafkaClusterRefTemplates.kafkaClusterRefCRWithTls(deploymentNamespace, clusterName),
101+
KroxyliciousVirtualKafkaClusterTemplates.defaultVirtualKafkaClusterCR(deploymentNamespace, clusterName, Constants.KROXYLICIOUS_PROXY_SIMPLE_NAME,
82102
clusterName, Constants.KROXYLICIOUS_INGRESS_CLUSTER_IP));
83103
}
84104

105+
private void createCertificateConfigMap(String namespace) {
106+
// wait for listeners to contain data
107+
var tlsListenerStatus = KafkaUtils.getKafkaListenerStatus("tls");
108+
109+
var cert = tlsListenerStatus.stream()
110+
.map(ListenerStatus::getCertificates)
111+
.findFirst().orElseThrow();
112+
113+
resourceManager.createResourceFromBuilder(KroxyliciousConfigMapTemplates.getClusterCaConfigMap(namespace, Constants.KROXYLICIOUS_TLS_CLIENT_CA_CERT,
114+
cert.get(0)));
115+
}
116+
85117
/**
86118
* Deploy port per broker plain with record encryption filter.
87119
*

‎kroxylicious-systemtests/src/main/java/io/kroxylicious/systemtests/templates/kroxylicious/KroxyliciousConfigMapTemplates.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,12 @@
66

77
package io.kroxylicious.systemtests.templates.kroxylicious;
88

9+
import java.util.Map;
10+
11+
import io.fabric8.kubernetes.api.model.ConfigMapBuilder;
12+
13+
import io.kroxylicious.systemtests.Constants;
14+
915
/**
1016
* The type Kroxylicious config templates.
1117
*/
@@ -38,4 +44,15 @@ public static String getDefaultExternalKroxyliciousConfigMap(String clusterExter
3844
"""
3945
.formatted(clusterExternalIP);
4046
}
47+
48+
public static ConfigMapBuilder getClusterCaConfigMap(String namespace, String name, String certificate) {
49+
// @formatter:off
50+
return new ConfigMapBuilder()
51+
.withNewMetadata()
52+
.withName(name)
53+
.withNamespace(namespace)
54+
.endMetadata()
55+
.withData(Map.of(Constants.KROXYLICIOUS_TLS_CA_NAME, certificate));
56+
// @formatter:on
57+
}
4158
}

‎kroxylicious-systemtests/src/main/java/io/kroxylicious/systemtests/templates/kroxylicious/KroxyliciousKafkaClusterRefTemplates.java‎

Lines changed: 48 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -6,31 +6,76 @@
66

77
package io.kroxylicious.systemtests.templates.kroxylicious;
88

9+
import io.strimzi.api.kafka.model.kafka.listener.ListenerStatus;
10+
11+
import io.kroxylicious.kubernetes.api.common.TrustAnchorRefBuilder;
912
import io.kroxylicious.kubernetes.api.v1alpha1.KafkaServiceBuilder;
13+
import io.kroxylicious.kubernetes.api.v1alpha1.kafkaservicespec.TlsBuilder;
1014
import io.kroxylicious.systemtests.Constants;
15+
import io.kroxylicious.systemtests.utils.KafkaUtils;
1116

1217
public class KroxyliciousKafkaClusterRefTemplates {
1318

1419
private KroxyliciousKafkaClusterRefTemplates() {
1520
}
1621

1722
/**
18-
* Default kafka cluster ref deployment.
23+
* Default kafka cluster ref CR.
1924
*
2025
* @param namespaceName the namespace name
2126
* @param clusterRefName the cluster ref name
2227
* @return the kafka service builder
2328
*/
24-
public static KafkaServiceBuilder defaultKafkaClusterRefDeployment(String namespaceName, String clusterRefName) {
29+
public static KafkaServiceBuilder defaultKafkaClusterRefCR(String namespaceName, String clusterRefName) {
2530
// @formatter:off
2631
return new KafkaServiceBuilder()
2732
.withNewMetadata()
2833
.withName(clusterRefName)
2934
.withNamespace(namespaceName)
3035
.endMetadata()
3136
.withNewSpec()
32-
.withBootstrapServers("%s-kafka-bootstrap.%s.svc.cluster.local:9092".formatted(clusterRefName, Constants.KAFKA_DEFAULT_NAMESPACE))
37+
.withBootstrapServers(getKafkaBootstrap("plain", clusterRefName))
3338
.endSpec();
3439
// @formatter:on
3540
}
41+
42+
/**
43+
* Kafka cluster ref CR with tls.
44+
*
45+
* @param namespaceName the namespace name
46+
* @param clusterRefName the cluster ref name
47+
* @return the kafka service builder
48+
*/
49+
public static KafkaServiceBuilder kafkaClusterRefCRWithTls(String namespaceName, String clusterRefName) {
50+
// @formatter:off
51+
return defaultKafkaClusterRefCR(namespaceName, clusterRefName)
52+
.editSpec()
53+
.withBootstrapServers(getKafkaBootstrap("tls", clusterRefName))
54+
.withTls(new TlsBuilder()
55+
.withTrustAnchorRef(new TrustAnchorRefBuilder()
56+
.withNewRef()
57+
.withName(Constants.KROXYLICIOUS_TLS_CLIENT_CA_CERT)
58+
.withKind(Constants.CONFIG_MAP)
59+
.endRef()
60+
.withKey(Constants.KROXYLICIOUS_TLS_CA_NAME)
61+
.build())
62+
.build())
63+
.endSpec();
64+
// @formatter:on
65+
}
66+
67+
private static String getKafkaBootstrap(String listenerStatusName, String clusterRefName) {
68+
// wait for listeners to contain data
69+
if (KafkaUtils.isKafkaUp(clusterRefName)) {
70+
var kafkaListenerStatus = KafkaUtils.getKafkaListenerStatus(listenerStatusName);
71+
72+
return kafkaListenerStatus.stream()
73+
.map(ListenerStatus::getBootstrapServers)
74+
.findFirst().orElseThrow();
75+
}
76+
else {
77+
// Some operator tests do not need kafka running so we can set a default value
78+
return String.format("%s-kafka-bootstrap.%s.svc.cluster.local:9092".formatted(clusterRefName, Constants.KAFKA_DEFAULT_NAMESPACE));
79+
}
80+
}
3681
}

‎kroxylicious-systemtests/src/main/java/io/kroxylicious/systemtests/templates/kroxylicious/KroxyliciousKafkaProxyIngressTemplates.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,14 @@ private KroxyliciousKafkaProxyIngressTemplates() {
1616
}
1717

1818
/**
19-
* Default kafka proxy ingress deployment.
19+
* Default kafka proxy ingress CR.
2020
*
2121
* @param namespaceName the namespace name
2222
* @param ingressName the ingress name
2323
* @param proxyName the name of the proxy to reference
2424
* @return the kafka proxy ingress builder
2525
*/
26-
public static KafkaProxyIngressBuilder defaultKafkaProxyIngressDeployment(String namespaceName, String ingressName, String proxyName) {
26+
public static KafkaProxyIngressBuilder defaultKafkaProxyIngressCR(String namespaceName, String ingressName, String proxyName) {
2727
// @formatter:off
2828
return new KafkaProxyIngressBuilder()
2929
.withNewMetadata()

‎kroxylicious-systemtests/src/main/java/io/kroxylicious/systemtests/templates/kroxylicious/KroxyliciousKafkaProxyTemplates.java‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -14,13 +14,13 @@ private KroxyliciousKafkaProxyTemplates() {
1414
}
1515

1616
/**
17-
* Default kafka proxy deployment.
17+
* Default kafka proxy CR.
1818
*
1919
* @param namespaceName the namespace name
2020
* @param name the name
2121
* @return the kafka proxy builder
2222
*/
23-
public static KafkaProxyBuilder defaultKafkaProxyDeployment(String namespaceName, String name) {
23+
public static KafkaProxyBuilder defaultKafkaProxyCR(String namespaceName, String name) {
2424
// @formatter:off
2525
return new KafkaProxyBuilder()
2626
.withNewMetadata()

‎kroxylicious-systemtests/src/main/java/io/kroxylicious/systemtests/templates/kroxylicious/KroxyliciousVirtualKafkaClusterTemplates.java‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@ public class KroxyliciousVirtualKafkaClusterTemplates {
2020
private KroxyliciousVirtualKafkaClusterTemplates() {
2121
}
2222

23-
private static VirtualKafkaClusterBuilder baseVirtualKafkaClusterDeployment(String namespaceName, String clusterName, String proxyName, String clusterRefName,
24-
String ingressName) {
23+
private static VirtualKafkaClusterBuilder baseVirtualKafkaClusterCR(String namespaceName, String clusterName, String proxyName, String clusterRefName,
24+
String ingressName) {
2525
// @formatter:off
2626
return new VirtualKafkaClusterBuilder()
2727
.withNewMetadata()
@@ -45,7 +45,7 @@ private static VirtualKafkaClusterBuilder baseVirtualKafkaClusterDeployment(Stri
4545
}
4646

4747
/**
48-
* Default virtual kafka cluster deployment.
48+
* Default virtual kafka cluster CR.
4949
*
5050
* @param namespaceName the namespace name
5151
* @param clusterName the cluster name
@@ -54,13 +54,13 @@ private static VirtualKafkaClusterBuilder baseVirtualKafkaClusterDeployment(Stri
5454
* @param ingressName the ingress name
5555
* @return the virtual kafka cluster builder
5656
*/
57-
public static VirtualKafkaClusterBuilder defaultVirtualKafkaClusterDeployment(String namespaceName, String clusterName, String proxyName, String clusterRefName,
58-
String ingressName) {
59-
return baseVirtualKafkaClusterDeployment(namespaceName, clusterName, proxyName, clusterRefName, ingressName);
57+
public static VirtualKafkaClusterBuilder defaultVirtualKafkaClusterCR(String namespaceName, String clusterName, String proxyName, String clusterRefName,
58+
String ingressName) {
59+
return baseVirtualKafkaClusterCR(namespaceName, clusterName, proxyName, clusterRefName, ingressName);
6060
}
6161

6262
/**
63-
* Default virtual kafka cluster deployment.
63+
* Default virtual kafka cluster CR.
6464
*
6565
* @param namespaceName the namespace name
6666
* @param clusterName the cluster name
@@ -70,9 +70,9 @@ public static VirtualKafkaClusterBuilder defaultVirtualKafkaClusterDeployment(St
7070
* @param filterNames the filter names
7171
* @return the virtual kafka cluster builder
7272
*/
73-
public static VirtualKafkaClusterBuilder virtualKafkaClusterWithFilterDeployment(String namespaceName, String clusterName, String proxyName, String clusterRefName,
74-
String ingressName, List<String> filterNames) {
75-
return baseVirtualKafkaClusterDeployment(namespaceName, clusterName, proxyName, clusterRefName, ingressName)
73+
public static VirtualKafkaClusterBuilder virtualKafkaClusterWithFilterCR(String namespaceName, String clusterName, String proxyName, String clusterRefName,
74+
String ingressName, List<String> filterNames) {
75+
return baseVirtualKafkaClusterCR(namespaceName, clusterName, proxyName, clusterRefName, ingressName)
7676
.editSpec()
7777
.addAllToFilterRefs(getFilterRefs(filterNames))
7878
.endSpec();

‎kroxylicious-systemtests/src/main/java/io/kroxylicious/systemtests/utils/KafkaUtils.java‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,17 +9,22 @@
99
import java.time.Duration;
1010
import java.util.List;
1111
import java.util.Objects;
12+
import java.util.Optional;
1213

1314
import org.slf4j.Logger;
1415
import org.slf4j.LoggerFactory;
1516

1617
import io.fabric8.kubernetes.api.model.Pod;
1718
import io.fabric8.kubernetes.api.model.batch.v1.Job;
19+
import io.strimzi.api.kafka.model.kafka.Kafka;
20+
import io.strimzi.api.kafka.model.kafka.KafkaStatus;
21+
import io.strimzi.api.kafka.model.kafka.listener.ListenerStatus;
1822

1923
import io.kroxylicious.systemtests.Constants;
2024
import io.kroxylicious.systemtests.executor.Exec;
2125
import io.kroxylicious.systemtests.executor.ExecResult;
2226
import io.kroxylicious.systemtests.k8s.exception.KubeClusterException;
27+
import io.kroxylicious.systemtests.resources.strimzi.KafkaType;
2328

2429
import static io.kroxylicious.systemtests.k8s.KubeClusterResource.kubeClient;
2530
import static org.awaitility.Awaitility.await;
@@ -135,4 +140,43 @@ private static String getPodUid(String deployNamespace, String podName) {
135140
return "";
136141
}
137142
}
143+
144+
/**
145+
* Gets kafka listener status.
146+
*
147+
* @param listenerStatusName the listener status name
148+
* @return the kafka listener status
149+
*/
150+
public static Optional<ListenerStatus> getKafkaListenerStatus(String listenerStatusName) {
151+
KafkaType type = new KafkaType();
152+
return await().atMost(Duration.ofSeconds(60))
153+
.pollInterval(Duration.ofMillis(200))
154+
.until(
155+
() -> type.getClient().inNamespace(Constants.KAFKA_DEFAULT_NAMESPACE)
156+
.list()
157+
.getItems()
158+
.stream()
159+
.findFirst()
160+
.map(Kafka::getStatus)
161+
.map(KafkaStatus::getListeners)
162+
.stream()
163+
.flatMap(List::stream)
164+
.filter(listenerStatus -> listenerStatus.getName().contains(listenerStatusName))
165+
.findFirst(),
166+
Optional::isPresent);
167+
}
168+
169+
/**
170+
* Is kafka up.
171+
*
172+
* @param clusterName the cluster name
173+
* @return the boolean
174+
*/
175+
public static boolean isKafkaUp(String clusterName) {
176+
List<Pod> kafkaPods = kubeClient().listPods(Constants.KAFKA_DEFAULT_NAMESPACE).stream()
177+
.filter(pod -> pod.getMetadata().getName().contains(clusterName))
178+
.toList();
179+
180+
return !kafkaPods.isEmpty();
181+
}
138182
}

0 commit comments

Comments
 (0)