ListenersValidator { static Set<String> validateAndGetErrorMessages(int replicas, List<GenericKafkaListener> listeners) { Set<String> errors = new HashSet<>(0); List<Integer> ports = getPorts(listeners); List<String> names = getNames(listeners); if (names.size() != listeners.size()) { errors.add("every listener needs to have a unique name"); } List<String> invalidNames = names.stream().filter(name -> !LISTENER_NAME_PATTERN.matcher(name).matches()).collect(Collectors.toList()); if (!invalidNames.isEmpty()) { errors.add("listener names " + invalidNames + " are invalid and do not match the pattern " + GenericKafkaListener.LISTENER_NAME_REGEX); } if (ports.size() != listeners.size()) { errors.add("every listener needs to have a unique port number"); } for (GenericKafkaListener listener : listeners) { validatePortNumbers(errors, listener); validateRouteAndIngressTlsOnly(errors, listener); validateTlsFeaturesOnNonTlsListener(errors, listener); validateOauth(errors, listener); if (listener.getConfiguration() != null) { validateServiceDnsDomain(errors, listener); validateIngressClass(errors, listener); validateExternalTrafficPolicy(errors, listener); validateLoadBalancerSourceRanges(errors, listener); validatePreferredAddressType(errors, listener); if (listener.getConfiguration().getBootstrap() != null) { validateBootstrapHost(errors, listener); validateBootstrapLoadBalancerIp(errors, listener); validateBootstrapNodePort(errors, listener); validateBootstrapDnsAnnotations(errors, listener); } if (listener.getConfiguration().getBrokers() != null) { for (GenericKafkaListenerConfigurationBroker broker : listener.getConfiguration().getBrokers()) { validateBrokerHost(errors, listener, broker); validateBrokerLoadBalancerIp(errors, listener, broker); validateBrokerNodePort(errors, listener, broker); validateBrokerDnsAnnotations(errors, listener, broker); } } } if (KafkaListenerType.INGRESS.equals(listener.getType())) { validateIngress(errors, replicas, listener); } } return errors; }  static void validate(int replicas, List<GenericKafkaListener> listeners); final static List<Integer> FORBIDDEN_PORTS; final static int LOWEST_ALLOWED_PORT_NUMBER; }
@Test public void testTlsAuthOnNonTlsListener() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("plain") .withPort(9092) .withType(KafkaListenerType.INTERNAL) .withTls(false) .withNewKafkaListenerAuthenticationTlsAuth() .endKafkaListenerAuthenticationTlsAuth() .build(); List<GenericKafkaListener> listeners = asList(listener1); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder("listener plain cannot use tls type authentication with disabled TLS encryption")); }
@Test public void testTlsCustomCertOnNonTlsListener() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("plain") .withPort(9092) .withType(KafkaListenerType.INTERNAL) .withTls(false) .withNewConfiguration() .withNewBrokerCertChainAndKey() .withCertificate("cert") .withKey("key") .withSecretName("secretName") .endBrokerCertChainAndKey() .endConfiguration() .build(); List<GenericKafkaListener> listeners = asList(listener1); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder("listener plain cannot configure custom TLS certificate with disabled TLS encryption")); }
@Test public void testInternalListener() { String name = "plain"; GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName(name) .withPort(9092) .withType(KafkaListenerType.INTERNAL) .withNewConfiguration() .withIngressClass("my-ingress") .withUseServiceDnsDomain(true) .withExternalTrafficPolicy(ExternalTrafficPolicy.LOCAL) .withPreferredNodePortAddressType(NodeAddressType.INTERNAL_DNS) .withLoadBalancerSourceRanges(asList("10.0.0.0/8", "130.211.204.1/32")) .withNewBootstrap() .withAlternativeNames(asList("my-name-1", "my-name-2")) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .endBootstrap() .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build()) .endConfiguration() .build(); List<GenericKafkaListener> listeners = asList(listener1); List<String> expectedErrors = asList( "listener " + name + " cannot configure ingressClass because it is not Ingress based listener", "listener " + name + " cannot configure externalTrafficPolicy because it is not LoadBalancer or NodePort based listener", "listener " + name + " cannot configure loadBalancerSourceRanges because it is not LoadBalancer based listener", "listener " + name + " cannot configure preferredAddressType because it is not NodePort based listener", "listener " + name + " cannot configure bootstrap.host because it is not Route ot Ingress based listener", "listener " + name + " cannot configure bootstrap.loadBalancerIP because it is not LoadBalancer based listener", "listener " + name + " cannot configure bootstrap.nodePort because it is not NodePort based listener", "listener " + name + " cannot configure bootstrap.dnsAnnotations because it is not LoadBalancer, NodePort or Ingress based listener", "listener " + name + " cannot configure brokers[].host because it is not Route ot Ingress based listener", "listener " + name + " cannot configure brokers[].loadBalancerIP because it is not LoadBalancer based listener", "listener " + name + " cannot configure brokers[].nodePort because it is not NodePort based listener", "listener " + name + " cannot configure brokers[].dnsAnnotations because it is not LoadBalancer, NodePort or Ingress based listener" ); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder(expectedErrors.toArray())); }
@Test public void testLoadBalancerListener() { String name = "lb"; GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName(name) .withPort(9092) .withType(KafkaListenerType.LOADBALANCER) .withNewConfiguration() .withIngressClass("my-ingress") .withUseServiceDnsDomain(true) .withExternalTrafficPolicy(ExternalTrafficPolicy.LOCAL) .withPreferredNodePortAddressType(NodeAddressType.INTERNAL_DNS) .withLoadBalancerSourceRanges(asList("10.0.0.0/8", "130.211.204.1/32")) .withNewBootstrap() .withAlternativeNames(asList("my-name-1", "my-name-2")) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .endBootstrap() .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build()) .endConfiguration() .build(); List<GenericKafkaListener> listeners = asList(listener1); List<String> expectedErrors = asList( "listener " + name + " cannot configure ingressClass because it is not Ingress based listener", "listener " + name + " cannot configure useServiceDnsDomain because it is not internal listener", "listener " + name + " cannot configure preferredAddressType because it is not NodePort based listener", "listener " + name + " cannot configure bootstrap.host because it is not Route ot Ingress based listener", "listener " + name + " cannot configure bootstrap.nodePort because it is not NodePort based listener", "listener " + name + " cannot configure brokers[].host because it is not Route ot Ingress based listener", "listener " + name + " cannot configure brokers[].nodePort because it is not NodePort based listener" ); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder(expectedErrors.toArray())); }
@Test public void testNodePortListener() { String name = "nodeport"; GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName(name) .withPort(9092) .withType(KafkaListenerType.NODEPORT) .withNewConfiguration() .withIngressClass("my-ingress") .withUseServiceDnsDomain(true) .withExternalTrafficPolicy(ExternalTrafficPolicy.LOCAL) .withPreferredNodePortAddressType(NodeAddressType.INTERNAL_DNS) .withLoadBalancerSourceRanges(asList("10.0.0.0/8", "130.211.204.1/32")) .withNewBootstrap() .withAlternativeNames(asList("my-name-1", "my-name-2")) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .endBootstrap() .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build()) .endConfiguration() .build(); List<GenericKafkaListener> listeners = asList(listener1); List<String> expectedErrors = asList( "listener " + name + " cannot configure ingressClass because it is not Ingress based listener", "listener " + name + " cannot configure useServiceDnsDomain because it is not internal listener", "listener " + name + " cannot configure loadBalancerSourceRanges because it is not LoadBalancer based listener", "listener " + name + " cannot configure bootstrap.host because it is not Route ot Ingress based listener", "listener " + name + " cannot configure bootstrap.loadBalancerIP because it is not LoadBalancer based listener", "listener " + name + " cannot configure brokers[].host because it is not Route ot Ingress based listener", "listener " + name + " cannot configure brokers[].loadBalancerIP because it is not LoadBalancer based listener" ); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder(expectedErrors.toArray())); }
@Test public void testRouteListenerWithoutTls() { String name = "route"; GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName(name) .withPort(9092) .withType(KafkaListenerType.ROUTE) .withTls(false) .build(); List<GenericKafkaListener> listeners = asList(listener1); List<String> expectedErrors = asList( "listener " + name + " is Route or Ingress type listener and requires enabled TLS encryption" ); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder(expectedErrors.toArray())); }
@Test public void testRouteListener() { String name = "route"; GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName(name) .withPort(9092) .withType(KafkaListenerType.ROUTE) .withTls(true) .withNewConfiguration() .withIngressClass("my-ingress") .withUseServiceDnsDomain(true) .withExternalTrafficPolicy(ExternalTrafficPolicy.LOCAL) .withPreferredNodePortAddressType(NodeAddressType.INTERNAL_DNS) .withLoadBalancerSourceRanges(asList("10.0.0.0/8", "130.211.204.1/32")) .withNewBootstrap() .withAlternativeNames(asList("my-name-1", "my-name-2")) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .endBootstrap() .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build()) .endConfiguration() .build(); List<GenericKafkaListener> listeners = asList(listener1); List<String> expectedErrors = asList( "listener " + name + " cannot configure ingressClass because it is not Ingress based listener", "listener " + name + " cannot configure useServiceDnsDomain because it is not internal listener", "listener " + name + " cannot configure externalTrafficPolicy because it is not LoadBalancer or NodePort based listener", "listener " + name + " cannot configure loadBalancerSourceRanges because it is not LoadBalancer based listener", "listener " + name + " cannot configure preferredAddressType because it is not NodePort based listener", "listener " + name + " cannot configure bootstrap.loadBalancerIP because it is not LoadBalancer based listener", "listener " + name + " cannot configure bootstrap.nodePort because it is not NodePort based listener", "listener " + name + " cannot configure bootstrap.dnsAnnotations because it is not LoadBalancer, NodePort or Ingress based listener", "listener " + name + " cannot configure brokers[].loadBalancerIP because it is not LoadBalancer based listener", "listener " + name + " cannot configure brokers[].nodePort because it is not NodePort based listener", "listener " + name + " cannot configure brokers[].dnsAnnotations because it is not LoadBalancer, NodePort or Ingress based listener" ); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder(expectedErrors.toArray())); }
@Test public void testIngressListenerWithoutTls() { String name = "ingress"; GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName(name) .withPort(9092) .withType(KafkaListenerType.INGRESS) .withTls(false) .withNewConfiguration() .withNewBootstrap() .withHost("my-host") .endBootstrap() .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .withHost("my-host") .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .withHost("my-host") .build()) .endConfiguration() .build(); List<GenericKafkaListener> listeners = asList(listener1); List<String> expectedErrors = asList( "listener " + name + " is Route or Ingress type listener and requires enabled TLS encryption" ); assertThat(ListenersValidator.validateAndGetErrorMessages(2, listeners), containsInAnyOrder(expectedErrors.toArray())); }
@Test public void testIngressListener() { String name = "ingress"; GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName(name) .withPort(9092) .withType(KafkaListenerType.INGRESS) .withTls(true) .withNewConfiguration() .withIngressClass("my-ingress") .withUseServiceDnsDomain(true) .withExternalTrafficPolicy(ExternalTrafficPolicy.LOCAL) .withPreferredNodePortAddressType(NodeAddressType.INTERNAL_DNS) .withLoadBalancerSourceRanges(asList("10.0.0.0/8", "130.211.204.1/32")) .withNewBootstrap() .withAlternativeNames(asList("my-name-1", "my-name-2")) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .endBootstrap() .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .withAdvertisedHost("advertised-host") .withAdvertisedPort(9092) .withLoadBalancerIP("130.211.204.1") .withNodePort(32189) .withHost("my-host") .withAnnotations(Collections.singletonMap("dns-anno", "dns-value")) .build()) .endConfiguration() .build(); List<GenericKafkaListener> listeners = asList(listener1); List<String> expectedErrors = asList( "listener " + name + " cannot configure useServiceDnsDomain because it is not internal listener", "listener " + name + " cannot configure externalTrafficPolicy because it is not LoadBalancer or NodePort based listener", "listener " + name + " cannot configure loadBalancerSourceRanges because it is not LoadBalancer based listener", "listener " + name + " cannot configure preferredAddressType because it is not NodePort based listener", "listener " + name + " cannot configure bootstrap.loadBalancerIP because it is not LoadBalancer based listener", "listener " + name + " cannot configure bootstrap.nodePort because it is not NodePort based listener", "listener " + name + " cannot configure brokers[].loadBalancerIP because it is not LoadBalancer based listener", "listener " + name + " cannot configure brokers[].nodePort because it is not NodePort based listener" ); assertThat(ListenersValidator.validateAndGetErrorMessages(2, listeners), containsInAnyOrder(expectedErrors.toArray())); }
@Test public void testIngressListenerHostNames() { GenericKafkaListener listener = new GenericKafkaListenerBuilder() .withName("ingress") .withPort(9092) .withType(KafkaListenerType.INGRESS) .withTls(true) .build(); assertThat(ListenersValidator.validateAndGetErrorMessages(2, asList(listener)), containsInAnyOrder("listener ingress is missing a configuration with host names which is required for Ingress based listeners")); listener.setConfiguration(new GenericKafkaListenerConfigurationBuilder() .withBrokers((List<GenericKafkaListenerConfigurationBroker>) null) .build()); assertThat(ListenersValidator.validateAndGetErrorMessages(2, asList(listener)), containsInAnyOrder("listener ingress is missing a bootstrap host name which is required for Ingress based listeners", "listener ingress is missing a broker configuration with host names which is required for Ingress based listeners")); listener.setConfiguration(new GenericKafkaListenerConfigurationBuilder() .withBootstrap(new GenericKafkaListenerConfigurationBootstrapBuilder() .build()) .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .build()) .build()); assertThat(ListenersValidator.validateAndGetErrorMessages(2, asList(listener)), containsInAnyOrder("listener ingress is missing a bootstrap host name which is required for Ingress based listeners", "listener ingress is missing a broker host name for broker with ID 0 which is required for Ingress based listeners", "listener ingress is missing a broker host name for broker with ID 1 which is required for Ingress based listeners")); listener.setConfiguration(new GenericKafkaListenerConfigurationBuilder() .withBootstrap(new GenericKafkaListenerConfigurationBootstrapBuilder() .withHost("bootstrap-host") .build()) .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .withHost("host-1") .build()) .build()); assertThat(ListenersValidator.validateAndGetErrorMessages(2, asList(listener)), containsInAnyOrder("listener ingress is missing a broker host name for broker with ID 0 which is required for Ingress based listeners")); listener.setConfiguration(new GenericKafkaListenerConfigurationBuilder() .withBootstrap(new GenericKafkaListenerConfigurationBootstrapBuilder() .withHost("bootstrap-host") .build()) .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .withHost("host-0") .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .withHost("host-1") .build()) .build()); assertThat(ListenersValidator.validateAndGetErrorMessages(2, asList(listener)), hasSize(0)); }
@Test public void testMinimalConfiguration() { GenericKafkaListener internal = new GenericKafkaListenerBuilder() .withName("internal") .withPort(9092) .withType(KafkaListenerType.INTERNAL) .build(); GenericKafkaListener route = new GenericKafkaListenerBuilder() .withName("route") .withPort(9093) .withType(KafkaListenerType.ROUTE) .withTls(true) .build(); GenericKafkaListener lb = new GenericKafkaListenerBuilder() .withName("lb") .withPort(9094) .withType(KafkaListenerType.LOADBALANCER) .build(); GenericKafkaListener np = new GenericKafkaListenerBuilder() .withName("np") .withPort(9095) .withType(KafkaListenerType.NODEPORT) .build(); GenericKafkaListener inq = new GenericKafkaListenerBuilder() .withName("inq") .withPort(9096) .withType(KafkaListenerType.INGRESS) .withTls(true) .withNewConfiguration() .withNewBootstrap() .withHost("my-host") .endBootstrap() .withBrokers(new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(0) .withHost("my-host") .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(1) .withHost("my-host") .build(), new GenericKafkaListenerConfigurationBrokerBuilder() .withBroker(2) .withHost("my-host") .build()) .endConfiguration() .build(); List<GenericKafkaListener> listeners = asList(internal, route, lb, np, inq); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), empty()); }
@Test public void testValidateDuplicatePorts() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("listener1") .withPort(9900) .withType(KafkaListenerType.INTERNAL) .build(); GenericKafkaListener listener2 = new GenericKafkaListenerBuilder() .withName("listener2") .withPort(9901) .withType(KafkaListenerType.INTERNAL) .build(); GenericKafkaListener listener3 = new GenericKafkaListenerBuilder() .withName("listener3") .withPort(9901) .withType(KafkaListenerType.INTERNAL) .build(); List<GenericKafkaListener> listeners = asList(listener1, listener2, listener3); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder("every listener needs to have a unique port number")); }
@Test public void testValidateForbiddenPortByRange() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("listener1") .withPort(9000) .withType(KafkaListenerType.INTERNAL) .build(); List<GenericKafkaListener> listeners = asList(listener1); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder("port 9000 is forbidden and cannot be used")); }
@Test public void testValidateForbiddenPortByException() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("listener1") .withPort(9404) .withType(KafkaListenerType.INTERNAL) .build(); List<GenericKafkaListener> listeners = asList(listener1); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder("port 9404 is forbidden and cannot be used")); }
@Test public void testValidateDuplicateNames() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("listener1") .withPort(9900) .withType(KafkaListenerType.INTERNAL) .build(); GenericKafkaListener listener2 = new GenericKafkaListenerBuilder() .withName("listener2") .withPort(9901) .withType(KafkaListenerType.INTERNAL) .build(); GenericKafkaListener listener3 = new GenericKafkaListenerBuilder() .withName("listener1") .withPort(9902) .withType(KafkaListenerType.INTERNAL) .build(); List<GenericKafkaListener> listeners = asList(listener1, listener2, listener3); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder("every listener needs to have a unique name")); }
@Test public void testInvalidNames() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("listener 1") .withPort(9900) .withType(KafkaListenerType.INTERNAL) .build(); GenericKafkaListener listener2 = new GenericKafkaListenerBuilder() .withName("LISTENER2") .withPort(9901) .withType(KafkaListenerType.INTERNAL) .build(); GenericKafkaListener listener3 = new GenericKafkaListenerBuilder() .withName("listener12345678901234567890") .withPort(9902) .withType(KafkaListenerType.INTERNAL) .build(); List<GenericKafkaListener> listeners = asList(listener1, listener2, listener3); assertThat(ListenersValidator.validateAndGetErrorMessages(3, listeners), containsInAnyOrder("listener names [listener 1, LISTENER2, listener12345678901234567890] are invalid and do not match the pattern ^[a-z0-9]{1,11}$")); }