### Question:
PulsarRecordsStorageConfiguration implements ApplicationContextInitializer<GenericApplicationContext> { @SneakyThrows PulsarClient createClient(PulsarProperties pulsarProperties) { var clientBuilder = PulsarClient.builder() .serviceUrl(pulsarProperties.getServiceUrl()); pulsarProperties.getTlsTrustCertsFilePath().ifPresent(clientBuilder::tlsTrustCertsFilePath); pulsarProperties.getAuthPluginClassName().ifPresent(authClass -> { try { clientBuilder.authentication(authClass, pulsarProperties.getAuthPluginParams()); } catch (PulsarClientException.UnsupportedAuthenticationException e) { throw new IllegalStateException(e); } }); return clientBuilder.build(); }  @Override void initialize(GenericApplicationContext applicationContext);  }### Answer:
@Test void shouldSupportTLS() throws Exception { try (var pulsar = new PulsarTlsContainer()) { pulsar.start(); var configuration = new PulsarRecordsStorageConfiguration(); var properties = new PulsarProperties(); properties.setServiceUrl(pulsar.getPulsarBrokerUrl()); var topicName = UUID.randomUUID().toString(); try (var client = configuration.createClient(properties)) { assertThatThrownBy(() -> client.getPartitionsForTopic(topicName).get(10, TimeUnit.SECONDS)) .hasCauseInstanceOf(PulsarClientException.class); } properties.setTlsTrustCertsFilePath(Optional.of(pulsar.getCaCert().toAbsolutePath().toString())); try (var client = configuration.createClient(properties)) { assertThat(client.getPartitionsForTopic(topicName).get(10, TimeUnit.SECONDS)) .isNotEmpty(); } } }
@Test void shouldSupportAuth() throws Exception { try (var pulsar = new PulsarTlsContainer()) { pulsar.withCopyFileToContainer(MountableFile.forClasspathResource(".htpasswd"), "/pulsar/conf/.htpasswd"); pulsar.withEnv("PULSAR_MEM", "-Dpulsar.auth.basic.conf=/pulsar/conf/.htpasswd"); pulsar.withEnv("superUserRoles", "super"); pulsar.withEnv("authenticationEnabled", "true"); pulsar.withEnv("authenticationProviders", "org.apache.pulsar.broker.authentication.AuthenticationProviderBasic"); pulsar.withEnv("brokerClientAuthenticationPlugin", "org.apache.pulsar.client.impl.auth.AuthenticationBasic"); pulsar.withEnv("brokerClientAuthenticationParameters", "{\"userId\":\"super\",\"password\":\"superpass\"}"); pulsar.start(); var configuration = new PulsarRecordsStorageConfiguration(); var topicName = UUID.randomUUID().toString(); var properties = new PulsarProperties(); properties.setServiceUrl(pulsar.getPulsarBrokerUrl()); properties.setTlsTrustCertsFilePath(Optional.of(pulsar.getCaCert().toAbsolutePath().toString())); try (var client = configuration.createClient(properties)) { assertThatThrownBy(() -> client.getPartitionsForTopic(topicName).get(10, TimeUnit.SECONDS)) .hasCauseInstanceOf(PulsarClientException.class); } properties.setAuthPluginClassName(Optional.of(AuthenticationBasic.class.getName())); properties.setAuthPluginParams(Map.of( "userId", "super", "password", UUID.randomUUID().toString() )); try (var client = configuration.createClient(properties)) { assertThatThrownBy(() -> client.getPartitionsForTopic(topicName).get(10, TimeUnit.SECONDS)) .hasCauseInstanceOf(PulsarClientException.class); } properties.setAuthPluginParams(Map.of( "userId", "super", "password", "superpass" )); try (var client = configuration.createClient(properties)) { assertThat(client.getPartitionsForTopic(topicName).get(10, TimeUnit.SECONDS)) .isNotEmpty(); } } }