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