Use authParamString to configure Pulsar authentication

Update `PulsarPropertiesMapper` to use JSON encoded parameters rather
than a `Map` since the `Map` method is deprecated in Pulsar. This
commit simply takes the auth params map and converts them to the
expected encoded JSON string of auth parameters.

See gh-38839

Author: onobc

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

@@ -19,6 +19,7 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Map;
+import java.util.TreeMap;
 import java.util.concurrent.TimeUnit;
 import java.util.function.BiConsumer;
 import java.util.function.Consumer;
@@ -29,6 +30,7 @@
 import org.apache.pulsar.client.api.ProducerBuilder;
 import org.apache.pulsar.client.api.PulsarClientException.UnsupportedAuthenticationException;
 import org.apache.pulsar.client.api.ReaderBuilder;
+import org.apache.pulsar.common.util.ObjectMapperFactory;
 
 import org.springframework.boot.context.properties.PropertyMapper;
 import org.springframework.pulsar.listener.PulsarContainerProperties;
@@ -71,13 +73,23 @@ void customizeAdminBuilder(PulsarAdminBuilder adminBuilder, PulsarConnectionDeta
 
 	private void customizeAuthentication(AuthenticationConsumer authentication,
 			PulsarProperties.Authentication properties) {
-		if (StringUtils.hasText(properties.getPluginClassName())) {
+		if (!StringUtils.hasText(properties.getPluginClassName())) {
+			return;
+		}
+		try {
+			// sort keys for testing and readability
+			Map<String, String> params = new TreeMap<>(properties.getParam());
+			String authParamString;
 			try {
-				authentication.accept(properties.getPluginClassName(), properties.getParam());
+				authParamString = ObjectMapperFactory.create().writeValueAsString(params);
 			}
-			catch (UnsupportedAuthenticationException ex) {
-				throw new IllegalStateException("Unable to configure Pulsar authentication", ex);
+			catch (Exception ex) {
+				throw new IllegalStateException("Could not convert auth parameters to encoded string", ex);
 			}
+			authentication.accept(properties.getPluginClassName(), authParamString);
+		}
+		catch (UnsupportedAuthenticationException ex) {
+			throw new IllegalStateException("Unable to configure Pulsar authentication", ex);
 		}
 	}
 
@@ -158,8 +170,7 @@ private Consumer<Duration> timeoutProperty(BiConsumer<Integer, TimeUnit> setter)
 
 	private interface AuthenticationConsumer {
 
-		void accept(String authPluginClassName, Map<String, String> authParams)
-				throws UnsupportedAuthenticationException;
+		void accept(String authPluginClassName, String authParamString) throws UnsupportedAuthenticationException;
 
 	}
 

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

@@ -73,12 +73,13 @@ void customizeClientBuilderWhenHasNoAuthentication() {
 	void customizeClientBuilderWhenHasAuthentication() throws UnsupportedAuthenticationException {
 		PulsarProperties properties = new PulsarProperties();
 		Map<String, String> params = Map.of("param", "name");
+		String authParamString = "{\"param\":\"name\"}";
 		properties.getClient().getAuthentication().setPluginClassName("myclass");
 		properties.getClient().getAuthentication().setParam(params);
 		ClientBuilder builder = mock(ClientBuilder.class);
 		new PulsarPropertiesMapper(properties).customizeClientBuilder(builder,
 				new PropertiesPulsarConnectionDetails(properties));
-		then(builder).should().authentication("myclass", params);
+		then(builder).should().authentication("myclass", authParamString);
 	}
 
 	@Test
@@ -112,12 +113,13 @@ void customizeAdminBuilderWhenHasNoAuthentication() {
 	void customizeAdminBuilderWhenHasAuthentication() throws UnsupportedAuthenticationException {
 		PulsarProperties properties = new PulsarProperties();
 		Map<String, String> params = Map.of("param", "name");
+		String authParamString = "{\"param\":\"name\"}";
 		properties.getAdmin().getAuthentication().setPluginClassName("myclass");
 		properties.getAdmin().getAuthentication().setParam(params);
 		PulsarAdminBuilder builder = mock(PulsarAdminBuilder.class);
 		new PulsarPropertiesMapper(properties).customizeAdminBuilder(builder,
 				new PropertiesPulsarConnectionDetails(properties));
-		then(builder).should().authentication("myclass", params);
+		then(builder).should().authentication("myclass", authParamString);
 	}
 
 	@Test