@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())); }
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; }
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; } }
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; }  }
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); }
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; }