PulsarAdministration now respects auth properties (#195)

* Add test to prove PulsarAdministration not respecting auth

* PulsarAdministration respects auth

* Fix condition in admin properties

spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java

@@ -51,6 +51,7 @@ import org.springframework.util.unit.DataSize;
  * @author Soby Chacko
  * @author Alexander Preuß
  * @author Christophe Bornet
+ * @author Chris Bono
  */
 @ConfigurationProperties(prefix = "spring.pulsar")
 public class PulsarProperties {
@@ -1609,9 +1610,9 @@ public class PulsarProperties {
 		}
 
 		public Map<String, Object> buildProperties() {
-			if (!StringUtils.hasText(this.getAuthParams()) && !CollectionUtils.isEmpty(this.getAuthentication())) {
+			if (StringUtils.hasText(this.getAuthParams()) && !CollectionUtils.isEmpty(this.getAuthentication())) {
 				throw new IllegalArgumentException(
-						"Cannot set both spring.pulsar.admin.authParams and spring.pulsar.admin.authentication.*");
+						"Cannot set both spring.pulsar.administration.authParams and spring.pulsar.administration.authentication.*");
 			}
 			PulsarProperties.Properties properties = new Properties();
 

spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2022 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.springframework.pulsar.autoconfigure;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.context.properties.bind.Bindable;
+import org.springframework.boot.context.properties.bind.Binder;
+import org.springframework.boot.context.properties.source.ConfigurationPropertySource;
+import org.springframework.boot.context.properties.source.MapConfigurationPropertySource;
+
+/**
+ * Unit tests for {@link PulsarProperties}.
+ *
+ * @author Chris Bono
+ */
+public class PulsarPropertiesTests {
+
+	private final PulsarProperties properties = new PulsarProperties();
+
+	private void bind(String name, String value) {
+		bind(Collections.singletonMap(name, value));
+	}
+
+	private void bind(Map<String, String> map) {
+		ConfigurationPropertySource source = new MapConfigurationPropertySource(map);
+		new Binder(source).bind("spring.pulsar", Bindable.ofInstance(this.properties));
+	}
+
+	@Nested
+	class AdminPropertiesTests {
+
+		private String authPluginClassName = "org.apache.pulsar.client.impl.auth.AuthenticationToken";
+
+		private String authParamsStr = "{\"token\":\"1234\"}";
+
+		private String authToken = "1234";
+
+		@Test
+		void authenticationUsingAuthParamsString() {
+			Map<String, String> props = new HashMap<>();
+			props.put("spring.pulsar.administration.auth-plugin-class-name",
+					"org.apache.pulsar.client.impl.auth.AuthenticationToken");
+			props.put("spring.pulsar.administration.auth-params", authParamsStr);
+			bind(props);
+			assertThat(properties.getAdministration().getAuthParams()).isEqualTo(authParamsStr);
+			assertThat(properties.getAdministration().getAuthPluginClassName()).isEqualTo(authPluginClassName);
+			Map<String, Object> adminProps = properties.buildAdminProperties();
+			assertThat(adminProps).containsEntry("authPluginClassName", authPluginClassName).containsEntry("authParams",
+					authParamsStr);
+		}
+
+		@Test
+		void authenticationUsingAuthenticationMap() {
+			Map<String, String> props = new HashMap<>();
+			props.put("spring.pulsar.administration.auth-plugin-class-name", authPluginClassName);
+			props.put("spring.pulsar.administration.authentication.token", authToken);
+			bind(props);
+			assertThat(properties.getAdministration().getAuthentication()).containsEntry("token", authToken);
+			assertThat(properties.getAdministration().getAuthPluginClassName()).isEqualTo(authPluginClassName);
+			Map<String, Object> adminProps = properties.buildAdminProperties();
+			assertThat(adminProps).containsEntry("authPluginClassName", authPluginClassName).containsEntry("authParams",
+					authParamsStr);
+		}
+
+		@Test
+		void authenticationNotAllowedUsingBothAuthParamsStringAndAuthenticationMap() {
+			Map<String, String> props = new HashMap<>();
+			props.put("spring.pulsar.administration.auth-plugin-class-name", authPluginClassName);
+			props.put("spring.pulsar.administration.auth-params", authParamsStr);
+			props.put("spring.pulsar.administration.authentication.token", authToken);
+			bind(props);
+			assertThatIllegalArgumentException().isThrownBy(() -> properties.buildAdminProperties())
+					.withMessageContaining(
+							"Cannot set both spring.pulsar.administration.authParams and spring.pulsar.administration.authentication.*");
+		}
+
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java

@@ -29,6 +29,7 @@ import org.apache.pulsar.client.admin.PulsarAdmin;
 import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
 
 import org.springframework.beans.BeansException;
 import org.springframework.beans.factory.SmartInitializingSingleton;
@@ -36,12 +37,14 @@ import org.springframework.context.ApplicationContext;
 import org.springframework.context.ApplicationContextAware;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.util.CollectionUtils;
+import org.springframework.util.StringUtils;
 
 /**
  * An administration class that delegates to {@link PulsarAdmin} to create and manage
  * topics defined in the application context.
  *
  * @author Alexander Preuß
+ * @author Chris Bono
  */
 public class PulsarAdministration
 		implements ApplicationContextAware, SmartInitializingSingleton, PulsarAdministrationOperations {
@@ -59,6 +62,19 @@ public class PulsarAdministration
 	 */
 	public PulsarAdministration(Map<String, Object> adminConfig) {
 		this.adminBuilder = PulsarAdmin.builder().loadConf(adminConfig);
+
+		// Workaround the fact that the PulsarAdminImpl does not attempt to construct the
+		// authentication from the config props
+		String authPluginClassName = (String) adminConfig.get("authPluginClassName");
+		String authParams = (String) adminConfig.get("authParams");
+		if (StringUtils.hasText(authPluginClassName) && StringUtils.hasText(authParams)) {
+			try {
+				this.adminBuilder.authentication(authPluginClassName, authParams);
+			}
+			catch (UnsupportedAuthenticationException ex) {
+				throw new RuntimeException("Unable to create admin auth: " + ex.getMessage(), ex);
+			}
+		}
 	}
 
 	/**

spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarAdministrationTests.java

@@ -18,13 +18,19 @@ package org.springframework.pulsar.core;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalStateException;
+import static org.assertj.core.api.InstanceOfAssertFactories.type;
 
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.apache.pulsar.client.admin.PulsarAdminBuilder;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.impl.auth.AuthenticationBasic;
+import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
 import org.junit.jupiter.api.Nested;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -40,6 +46,7 @@ import org.springframework.test.context.junit.jupiter.SpringExtension;
  * Tests for {@link PulsarAdministration}.
  *
  * @author Alexander Preuß
+ * @author Chris Bono
  */
 @ExtendWith(SpringExtension.class)
 @ContextConfiguration
@@ -68,6 +75,19 @@ public class PulsarAdministrationTests implements PulsarTestContainerSupport {
 		assertThat(pulsarAdminClient.topics().getList(NAMESPACE)).containsAll(expectedTopics);
 	}
 
+	@Test
+	void constructorRespectsAuthenticationProps() {
+		Map<String, Object> props = new HashMap<>();
+		props.put("authPluginClassName", "org.apache.pulsar.client.impl.auth.AuthenticationBasic");
+		props.put("authParams", "{\"userId\":\"foo\", \"password\":\"bar\"}");
+		PulsarAdministration admin = new PulsarAdministration(props);
+
+		assertThat(admin).extracting("adminBuilder").asInstanceOf(type(PulsarAdminBuilder.class)).extracting("conf")
+				.asInstanceOf(type(ClientConfigurationData.class))
+				.extracting(ClientConfigurationData::getAuthentication).isInstanceOf(AuthenticationBasic.class)
+				.hasFieldOrPropertyWithValue("userId", "foo").hasFieldOrPropertyWithValue("password", "bar");
+	}
+
 	@Configuration(proxyBeanMethods = false)
 	static class AdminConfiguration {
 

src/checkstyle/checkstyle.xml

@@ -82,6 +82,7 @@
 		<module name="AvoidStaticImport">
 			<property name="excludes"
 					  value="org.assertj.core.api.Assertions.*,
+				org.assertj.core.api.InstanceOfAssertFactories.type,
 				org.awaitility.Awaitility.*,
 				org.junit.jupiter.api.Assertions.*,
 				org.junit.jupiter.params.provider.Arguments.*,