ListenersValidator { public static void validate(int replicas, List<GenericKafkaListener> listeners) throws InvalidResourceException { Set<String> errors = validateAndGetErrorMessages(replicas, listeners); if (!errors.isEmpty()) { LOG.error("Listener configuration is not valid: {}", errors); throw new InvalidResourceException("Listener configuration is not valid: " + 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 testValidateOldListener() { KafkaListeners oldListeners = new KafkaListenersBuilder() .withNewPlain() .withAuth(new KafkaListenerAuthenticationScramSha512()) .endPlain() .withNewTls() .withAuth(new KafkaListenerAuthenticationTls()) .endTls() .withNewKafkaListenerExternalRoute() .withAuth(new KafkaListenerAuthenticationTls()) .endKafkaListenerExternalRoute() .build(); List<GenericKafkaListener> newListeners = ListenersConvertor.convertToNewFormat(oldListeners); ListenersValidator.validate(3, newListeners); }
@Test public void testValidateNewListeners() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("listener1") .withPort(9900) .withType(KafkaListenerType.INTERNAL) .build(); GenericKafkaListener listener2 = new GenericKafkaListenerBuilder() .withName("listener2") .withPort(9901) .withType(KafkaListenerType.INTERNAL) .build(); List<GenericKafkaListener> listeners = asList(listener1, listener2); ListenersValidator.validate(3, listeners); }
@Test public void testValidateThrowsException() { GenericKafkaListener listener1 = new GenericKafkaListenerBuilder() .withName("listener1") .withPort(9900) .withType(KafkaListenerType.INTERNAL) .build(); GenericKafkaListener listener2 = new GenericKafkaListenerBuilder() .withName("listener2") .withPort(9900) .withType(KafkaListenerType.INTERNAL) .build(); List<GenericKafkaListener> listeners = asList(listener1, listener2); Exception exception = assertThrows(InvalidResourceException.class, () -> ListenersValidator.validate(3, listeners)); assertThat(exception.getMessage(), containsString("every listener needs to have a unique port number")); }