ListenersUtils { public static String brokerAdvertisedHost(GenericKafkaListener listener, int pod) { if (listener.getConfiguration() != null && listener.getConfiguration().getBrokers() != null) { return listener.getConfiguration().getBrokers().stream() .filter(broker -> broker != null && broker.getBroker() != null && broker.getBroker() == pod && broker.getAdvertisedHost() != null) .map(GenericKafkaListenerConfigurationBroker::getAdvertisedHost) .findAny() .orElse(null); } else { return null; } }  static boolean hasListenerWithOAuth(List<GenericKafkaListener> listeners); static List<GenericKafkaListener> internalListeners(List<GenericKafkaListener> listeners); static List<GenericKafkaListener> externalListeners(List<GenericKafkaListener> listeners); static List<GenericKafkaListener> routeListeners(List<GenericKafkaListener> listeners); static List<GenericKafkaListener> loadBalancerListeners(List<GenericKafkaListener> listeners); static List<GenericKafkaListener> nodePortListeners(List<GenericKafkaListener> listeners); static List<GenericKafkaListener> ingressListeners(List<GenericKafkaListener> listeners); static boolean hasExternalListener(List<GenericKafkaListener> listeners); static boolean hasRouteListener(List<GenericKafkaListener> listeners); static boolean hasLoadBalancerListener(List<GenericKafkaListener> listeners); static boolean hasNodePortListener(List<GenericKafkaListener> listeners); static boolean hasIngressListener(List<GenericKafkaListener> listeners); static List<String> alternativeNames(List<GenericKafkaListener> listeners); static String identifier(GenericKafkaListener listener); static String envVarIdentifier(GenericKafkaListener listener); static String backwardsCompatiblePortName(GenericKafkaListener listener); static String backwardsCompatibleBootstrapServiceName(String clusterName, GenericKafkaListener listener); static String backwardsCompatibleBootstrapRouteOrIngressName(String clusterName, GenericKafkaListener listener); static String backwardsCompatibleBrokerServiceName(String clusterName, int pod, GenericKafkaListener listener); static Integer bootstrapNodePort(GenericKafkaListener listener); static Integer brokerNodePort(GenericKafkaListener listener, int pod); static String bootstrapLoadBalancerIP(GenericKafkaListener listener); static String brokerLoadBalancerIP(GenericKafkaListener listener, int pod); static Map<String, String> bootstrapAnnotations(GenericKafkaListener listener); static Map<String, String> brokerAnnotations(GenericKafkaListener listener, int pod); static String bootstrapHost(GenericKafkaListener listener); static String brokerHost(GenericKafkaListener listener, int pod); static String brokerAdvertisedHost(GenericKafkaListener listener, int pod); static Integer brokerAdvertisedPort(GenericKafkaListener listener, int pod); static List<String> loadBalancerSourceRanges(GenericKafkaListener listener); static ExternalTrafficPolicy externalTrafficPolicy(GenericKafkaListener listener); static NodeAddressType preferredNodeAddressType(GenericKafkaListener listener); static String ingressClass(GenericKafkaListener listener); static String serviceType(GenericKafkaListener listener);  }
@Test public void testBrokerAdvertisedHost() { assertThat(ListenersUtils.brokerAdvertisedHost(newLoadBalancer, 1), is(nullValue())); assertThat(ListenersUtils.brokerAdvertisedHost(oldExternal, 0), is(nullValue())); assertThat(ListenersUtils.brokerAdvertisedHost(newLoadBalancer2, 0), is("advertised-host-1")); assertThat(ListenersUtils.brokerAdvertisedHost(newLoadBalancer2, 1), is("advertised-host-2")); assertThat(ListenersUtils.brokerAdvertisedHost(newLoadBalancer2, 2), is(nullValue())); assertThat(ListenersUtils.brokerAdvertisedHost(oldPlain, 1), is(nullValue())); assertThat(ListenersUtils.brokerAdvertisedHost(newTls, 1), is("advertised-host")); assertThat(ListenersUtils.brokerAdvertisedHost(newNodePort, 1), is(nullValue())); assertThat(ListenersUtils.brokerAdvertisedHost(newNodePort3, 1), is(nullValue())); }