ListenersConvertor { private static void convertNodePortListener(GenericKafkaListener newListener, KafkaListenerExternalNodePort oldListener) { newListener.setType(KafkaListenerType.NODEPORT); newListener.setTls(oldListener.isTls()); if (oldListener.getConfiguration() == null && oldListener.getOverrides() == null) { return; } GenericKafkaListenerConfiguration configuration = new GenericKafkaListenerConfiguration(); if (oldListener.getConfiguration() != null) { configuration.setBrokerCertChainAndKey(oldListener.getConfiguration().getBrokerCertChainAndKey()); configuration.setPreferredNodePortAddressType(oldListener.getConfiguration().getPreferredAddressType()); } if (oldListener.getOverrides() != null) { if (oldListener.getOverrides().getBootstrap() != null) { GenericKafkaListenerConfigurationBootstrap bootstrapConfiguration = new GenericKafkaListenerConfigurationBootstrap(); bootstrapConfiguration.setNodePort(oldListener.getOverrides().getBootstrap().getNodePort()); bootstrapConfiguration.setAnnotations(oldListener.getOverrides().getBootstrap().getDnsAnnotations()); if (oldListener.getOverrides().getBootstrap().getAddress() != null) { bootstrapConfiguration.setAlternativeNames(Collections.singletonList(oldListener.getOverrides().getBootstrap().getAddress())); } configuration.setBootstrap(bootstrapConfiguration); } if (oldListener.getOverrides().getBrokers() != null) { List<GenericKafkaListenerConfigurationBroker> brokers = new ArrayList<>(oldListener.getOverrides().getBrokers().size()); for (NodePortListenerBrokerOverride oldBroker : oldListener.getOverrides().getBrokers()) { GenericKafkaListenerConfigurationBroker brokerConfiguration = new GenericKafkaListenerConfigurationBroker(); brokerConfiguration.setBroker(oldBroker.getBroker()); brokerConfiguration.setNodePort(oldBroker.getNodePort()); brokerConfiguration.setAnnotations(oldBroker.getDnsAnnotations()); brokerConfiguration.setAdvertisedHost(oldBroker.getAdvertisedHost()); brokerConfiguration.setAdvertisedPort(oldBroker.getAdvertisedPort()); brokers.add(brokerConfiguration); } configuration.setBrokers(brokers); } } newListener.setConfiguration(configuration); }  @SuppressWarnings("deprecation") static List<GenericKafkaListener> convertToNewFormat(KafkaListeners oldListeners);  }
@Test public void testConvertNodePortListener() { KafkaListenerExternal oldListener = new KafkaListenerExternalNodePortBuilder() .withAuth(new KafkaListenerAuthenticationScramSha512()) .withNetworkPolicyPeers(NETWORK_POLICY_PEER_1, NETWORK_POLICY_PEER_2) .withTls(false) .withNewConfiguration() .withNewBrokerCertChainAndKey() .withCertificate("cert") .withKey("key") .withSecretName("secretName") .endBrokerCertChainAndKey() .withPreferredAddressType(NodeAddressType.EXTERNAL_IP) .endConfiguration() .withNewOverrides() .withNewBootstrap() .withNodePort(31000) .withAddress("my-bootstrap-dns") .withDnsAnnotations(Collections.singletonMap("anno", "my-dns")) .endBootstrap() .withBrokers( new NodePortListenerBrokerOverrideBuilder() .withBroker(0) .withAdvertisedHost("my-advertised-host-0") .withAdvertisedPort(1234) .withNodePort(32000) .withDnsAnnotations(Collections.singletonMap("anno", "my-dns-0")) .build(), new NodePortListenerBrokerOverrideBuilder() .withBroker(1) .withAdvertisedHost("my-advertised-host-1") .withAdvertisedPort(1234) .withNodePort(32001) .withDnsAnnotations(Collections.singletonMap("anno", "my-dns-1")) .build() ) .endOverrides() .build(); GenericKafkaListener newListener = ListenersConvertor.convertExternalListener(oldListener); assertThat(newListener.getName(), is("external")); assertThat(newListener.getPort(), is(9094)); assertThat(newListener.getType(), is(KafkaListenerType.NODEPORT)); assertThat(newListener.getAuth().getType(), is(KafkaListenerAuthenticationScramSha512.SCRAM_SHA_512)); assertThat(newListener.isTls(), is(false)); assertThat(newListener.getNetworkPolicyPeers().size(), is(2)); assertThat(newListener.getNetworkPolicyPeers(), hasItems(NETWORK_POLICY_PEER_1, NETWORK_POLICY_PEER_2)); assertThat(newListener.getConfiguration(), is(notNullValue())); assertThat(newListener.getConfiguration().getBrokerCertChainAndKey().getCertificate(), is("cert")); assertThat(newListener.getConfiguration().getBrokerCertChainAndKey().getKey(), is("key")); assertThat(newListener.getConfiguration().getBrokerCertChainAndKey().getSecretName(), is("secretName")); assertThat(newListener.getConfiguration().getPreferredNodePortAddressType(), is(NodeAddressType.EXTERNAL_IP)); assertThat(newListener.getConfiguration().getBootstrap().getNodePort(), is(31000)); assertThat(newListener.getConfiguration().getBootstrap().getAlternativeNames().size(), is(1)); assertThat(newListener.getConfiguration().getBootstrap().getAlternativeNames().get(0), is("my-bootstrap-dns")); assertThat(newListener.getConfiguration().getBootstrap().getAnnotations().get("anno"), is("my-dns")); assertThat(newListener.getConfiguration().getBrokers().size(), is(2)); assertThat(newListener.getConfiguration().getBrokers().get(0).getBroker(), is(0)); assertThat(newListener.getConfiguration().getBrokers().get(0).getAdvertisedHost(), is("my-advertised-host-0")); assertThat(newListener.getConfiguration().getBrokers().get(0).getAdvertisedPort(), is(1234)); assertThat(newListener.getConfiguration().getBrokers().get(0).getNodePort(), is(32000)); assertThat(newListener.getConfiguration().getBrokers().get(0).getAnnotations().get("anno"), is("my-dns-0")); assertThat(newListener.getConfiguration().getBrokers().get(1).getBroker(), is(1)); assertThat(newListener.getConfiguration().getBrokers().get(1).getAdvertisedHost(), is("my-advertised-host-1")); assertThat(newListener.getConfiguration().getBrokers().get(1).getAdvertisedPort(), is(1234)); assertThat(newListener.getConfiguration().getBrokers().get(1).getNodePort(), is(32001)); assertThat(newListener.getConfiguration().getBrokers().get(1).getAnnotations().get("anno"), is("my-dns-1")); }