Merge pull request #37197 from onobc

* pr/37197:
  Polish 'Add Pulsar ConnectionDetails support'
  Add Pulsar ConnectionDetails support

Closes gh-37197

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2012-2023 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.boot.autoconfigure.pulsar;
+
+/**
+ * Adapts {@link PulsarProperties} to {@link PulsarConnectionDetails}.
+ *
+ * @author Chris Bono
+ */
+class PropertiesPulsarConnectionDetails implements PulsarConnectionDetails {
+
+	private final PulsarProperties pulsarProperties;
+
+	PropertiesPulsarConnectionDetails(PulsarProperties pulsarProperties) {
+		this.pulsarProperties = pulsarProperties;
+	}
+
+	@Override
+	public String getBrokerUrl() {
+		return this.pulsarProperties.getClient().getServiceUrl();
+	}
+
+	@Override
+	public String getAdminUrl() {
+		return this.pulsarProperties.getAdmin().getServiceUrl();
+	}
+
+}

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

@@ -71,11 +71,18 @@ class PulsarConfiguration {
 		this.propertiesMapper = new PulsarPropertiesMapper(properties);
 	}
 
+	@Bean
+	@ConditionalOnMissingBean(PulsarConnectionDetails.class)
+	PropertiesPulsarConnectionDetails pulsarConnectionDetails() {
+		return new PropertiesPulsarConnectionDetails(this.properties);
+	}
+
 	@Bean
 	@ConditionalOnMissingBean(PulsarClientFactory.class)
-	DefaultPulsarClientFactory pulsarClientFactory(ObjectProvider<PulsarClientBuilderCustomizer> customizersProvider) {
+	DefaultPulsarClientFactory pulsarClientFactory(PulsarConnectionDetails connectionDetails,
+			ObjectProvider<PulsarClientBuilderCustomizer> customizersProvider) {
 		List<PulsarClientBuilderCustomizer> allCustomizers = new ArrayList<>();
-		allCustomizers.add(this.propertiesMapper::customizeClientBuilder);
+		allCustomizers.add((builder) -> this.propertiesMapper.customizeClientBuilder(builder, connectionDetails));
 		allCustomizers.addAll(customizersProvider.orderedStream().toList());
 		DefaultPulsarClientFactory clientFactory = new DefaultPulsarClientFactory(
 				(clientBuilder) -> applyClientBuilderCustomizers(allCustomizers, clientBuilder));
@@ -95,10 +102,10 @@ class PulsarConfiguration {
 
 	@Bean
 	@ConditionalOnMissingBean
-	PulsarAdministration pulsarAdministration(
+	PulsarAdministration pulsarAdministration(PulsarConnectionDetails connectionDetails,
 			ObjectProvider<PulsarAdminBuilderCustomizer> pulsarAdminBuilderCustomizers) {
 		List<PulsarAdminBuilderCustomizer> allCustomizers = new ArrayList<>();
-		allCustomizers.add(this.propertiesMapper::customizeAdminBuilder);
+		allCustomizers.add((builder) -> this.propertiesMapper.customizeAdminBuilder(builder, connectionDetails));
 		allCustomizers.addAll(pulsarAdminBuilderCustomizers.orderedStream().toList());
 		return new PulsarAdministration((adminBuilder) -> applyAdminBuilderCustomizers(allCustomizers, adminBuilder));
 	}

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

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2012-2023 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.boot.autoconfigure.pulsar;
+
+import org.springframework.boot.autoconfigure.service.connection.ConnectionDetails;
+
+/**
+ * Details required to establish a connection to a Pulsar service.
+ *
+ * @author Chris Bono
+ * @since 3.2.0
+ */
+public interface PulsarConnectionDetails extends ConnectionDetails {
+
+	/**
+	 * URL used to connect to the broker.
+	 * @return the service URL
+	 */
+	String getBrokerUrl();
+
+	/**
+	 * URL user to connect to the admin endpoint.
+	 * @return the admin URL
+	 */
+	String getAdminUrl();
+
+}

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

@@ -49,20 +49,20 @@ final class PulsarPropertiesMapper {
 		this.properties = properties;
 	}
 
-	void customizeClientBuilder(ClientBuilder clientBuilder) {
+	void customizeClientBuilder(ClientBuilder clientBuilder, PulsarConnectionDetails connectionDetails) {
 		PulsarProperties.Client properties = this.properties.getClient();
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
-		map.from(properties::getServiceUrl).to(clientBuilder::serviceUrl);
+		map.from(connectionDetails::getBrokerUrl).to(clientBuilder::serviceUrl);
 		map.from(properties::getConnectionTimeout).to(timeoutProperty(clientBuilder::connectionTimeout));
 		map.from(properties::getOperationTimeout).to(timeoutProperty(clientBuilder::operationTimeout));
 		map.from(properties::getLookupTimeout).to(timeoutProperty(clientBuilder::lookupTimeout));
 		customizeAuthentication(clientBuilder::authentication, properties.getAuthentication());
 	}
 
-	void customizeAdminBuilder(PulsarAdminBuilder adminBuilder) {
+	void customizeAdminBuilder(PulsarAdminBuilder adminBuilder, PulsarConnectionDetails connectionDetails) {
 		PulsarProperties.Admin properties = this.properties.getAdmin();
 		PropertyMapper map = PropertyMapper.get().alwaysApplyingWhenNonNull();
-		map.from(properties::getServiceUrl).to(adminBuilder::serviceHttpUrl);
+		map.from(connectionDetails::getAdminUrl).to(adminBuilder::serviceHttpUrl);
 		map.from(properties::getConnectionTimeout).to(timeoutProperty(adminBuilder::connectionTimeout));
 		map.from(properties::getReadTimeout).to(timeoutProperty(adminBuilder::readTimeout));
 		map.from(properties::getRequestTimeout).to(timeoutProperty(adminBuilder::requestTimeout));

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

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2012-2023 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.boot.autoconfigure.pulsar;
+
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link PropertiesPulsarConnectionDetails}.
+ *
+ * @author Chris Bono
+ */
+class PropertiesPulsarConnectionDetailsTests {
+
+	@Test
+	void getClientServiceUrlReturnsValueFromProperties() {
+		PulsarProperties properties = new PulsarProperties();
+		properties.getClient().setServiceUrl("foo");
+		PulsarConnectionDetails connectionDetails = new PropertiesPulsarConnectionDetails(properties);
+		assertThat(connectionDetails.getBrokerUrl()).isEqualTo("foo");
+	}
+
+	@Test
+	void getAdminServiceHttpUrlReturnsValueFromProperties() {
+		PulsarProperties properties = new PulsarProperties();
+		properties.getAdmin().setServiceUrl("foo");
+		PulsarConnectionDetails connectionDetails = new PropertiesPulsarConnectionDetails(properties);
+		assertThat(connectionDetails.getAdminUrl()).isEqualTo("foo");
+	}
+
+}

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

@@ -114,6 +114,7 @@ class PulsarAutoConfigurationTests {
 	@Test
 	void autoConfiguresBeans() {
 		this.contextRunner.run((context) -> assertThat(context).hasSingleBean(PulsarConfiguration.class)
+			.hasSingleBean(PulsarConnectionDetails.class)
 			.hasSingleBean(DefaultPulsarClientFactory.class)
 			.hasSingleBean(PulsarClient.class)
 			.hasSingleBean(PulsarAdministration.class)

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

@@ -51,6 +51,7 @@ import org.springframework.pulsar.function.PulsarFunctionAdministration;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.entry;
+import static org.mockito.BDDMockito.given;
 import static org.mockito.Mockito.mock;
 
 /**
@@ -67,6 +68,15 @@ class PulsarConfigurationTests {
 		.withConfiguration(AutoConfigurations.of(PulsarConfiguration.class))
 		.withBean(PulsarClient.class, () -> mock(PulsarClient.class));
 
+	@Test
+	void whenHasUserDefinedConnectionDetailsBeanDoesNotAutoConfigureBean() {
+		PulsarConnectionDetails customConnectionDetails = mock(PulsarConnectionDetails.class);
+		this.contextRunner
+			.withBean("customPulsarConnectionDetails", PulsarConnectionDetails.class, () -> customConnectionDetails)
+			.run((context) -> assertThat(context).getBean(PulsarConnectionDetails.class)
+				.isSameAs(customConnectionDetails));
+	}
+
 	@Nested
 	class ClientTests {
 
@@ -88,15 +98,18 @@ class PulsarConfigurationTests {
 
 		@Test
 		void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
+			PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
+			given(connectionDetails.getBrokerUrl()).willReturn("connectiondetails");
 			PulsarConfigurationTests.this.contextRunner
 				.withUserConfiguration(PulsarClientBuilderCustomizersConfig.class)
-				.withPropertyValues("spring.pulsar.client.service-url=fromPropsCustomizer")
+				.withBean(PulsarConnectionDetails.class, () -> connectionDetails)
+				.withPropertyValues("spring.pulsar.client.service-url=properties")
 				.run((context) -> {
 					DefaultPulsarClientFactory clientFactory = context.getBean(DefaultPulsarClientFactory.class);
 					Customizers<PulsarClientBuilderCustomizer, ClientBuilder> customizers = Customizers
 						.of(ClientBuilder.class, PulsarClientBuilderCustomizer::customize);
 					assertThat(customizers.fromField(clientFactory, "customizer")).callsInOrder(
-							ClientBuilder::serviceUrl, "fromPropsCustomizer", "fromCustomizer1", "fromCustomizer2");
+							ClientBuilder::serviceUrl, "connectiondetails", "fromCustomizer1", "fromCustomizer2");
 				});
 		}
 
@@ -135,14 +148,17 @@ class PulsarConfigurationTests {
 
 		@Test
 		void whenHasUserDefinedCustomizersAppliesInCorrectOrder() {
+			PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
+			given(connectionDetails.getAdminUrl()).willReturn("connectiondetails");
 			this.contextRunner.withUserConfiguration(PulsarAdminBuilderCustomizersConfig.class)
-				.withPropertyValues("spring.pulsar.admin.service-url=fromPropsCustomizer")
+				.withBean(PulsarConnectionDetails.class, () -> connectionDetails)
+				.withPropertyValues("spring.pulsar.admin.service-url=property")
 				.run((context) -> {
 					PulsarAdministration pulsarAdmin = context.getBean(PulsarAdministration.class);
 					Customizers<PulsarAdminBuilderCustomizer, PulsarAdminBuilder> customizers = Customizers
 						.of(PulsarAdminBuilder.class, PulsarAdminBuilderCustomizer::customize);
 					assertThat(customizers.fromField(pulsarAdmin, "adminCustomizers")).callsInOrder(
-							PulsarAdminBuilder::serviceHttpUrl, "fromPropsCustomizer", "fromCustomizer1",
+							PulsarAdminBuilder::serviceHttpUrl, "connectiondetails", "fromCustomizer1",
 							"fromCustomizer2");
 				});
 		}

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

@@ -41,6 +41,7 @@ import org.springframework.boot.autoconfigure.pulsar.PulsarProperties.Consumer;
 import org.springframework.pulsar.listener.PulsarContainerProperties;
 
 import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.BDDMockito.given;
 import static org.mockito.BDDMockito.then;
 import static org.mockito.Mockito.mock;
 
@@ -60,7 +61,8 @@ class PulsarPropertiesMapperTests {
 		properties.getClient().setOperationTimeout(Duration.ofSeconds(2));
 		properties.getClient().setLookupTimeout(Duration.ofSeconds(3));
 		ClientBuilder builder = mock(ClientBuilder.class);
-		new PulsarPropertiesMapper(properties).customizeClientBuilder(builder);
+		new PulsarPropertiesMapper(properties).customizeClientBuilder(builder,
+				new PropertiesPulsarConnectionDetails(properties));
 		then(builder).should().serviceUrl("https://example.com");
 		then(builder).should().connectionTimeout(1000, TimeUnit.MILLISECONDS);
 		then(builder).should().operationTimeout(2000, TimeUnit.MILLISECONDS);
@@ -74,10 +76,22 @@ class PulsarPropertiesMapperTests {
 		properties.getClient().getAuthentication().setPluginClassName("myclass");
 		properties.getClient().getAuthentication().setParam(params);
 		ClientBuilder builder = mock(ClientBuilder.class);
-		new PulsarPropertiesMapper(properties).customizeClientBuilder(builder);
+		new PulsarPropertiesMapper(properties).customizeClientBuilder(builder,
+				new PropertiesPulsarConnectionDetails(properties));
 		then(builder).should().authentication("myclass", params);
 	}
 
+	@Test
+	void customizeClientBuilderWhenHasConnectionDetails() {
+		PulsarProperties properties = new PulsarProperties();
+		properties.getClient().setServiceUrl("https://ignored.example.com");
+		ClientBuilder builder = mock(ClientBuilder.class);
+		PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
+		given(connectionDetails.getBrokerUrl()).willReturn("https://used.example.com");
+		new PulsarPropertiesMapper(properties).customizeClientBuilder(builder, connectionDetails);
+		then(builder).should().serviceUrl("https://used.example.com");
+	}
+
 	@Test
 	void customizeAdminBuilderWhenHasNoAuthentication() {
 		PulsarProperties properties = new PulsarProperties();
@@ -86,7 +100,8 @@ class PulsarPropertiesMapperTests {
 		properties.getAdmin().setReadTimeout(Duration.ofSeconds(2));
 		properties.getAdmin().setRequestTimeout(Duration.ofSeconds(3));
 		PulsarAdminBuilder builder = mock(PulsarAdminBuilder.class);
-		new PulsarPropertiesMapper(properties).customizeAdminBuilder(builder);
+		new PulsarPropertiesMapper(properties).customizeAdminBuilder(builder,
+				new PropertiesPulsarConnectionDetails(properties));
 		then(builder).should().serviceHttpUrl("https://example.com");
 		then(builder).should().connectionTimeout(1000, TimeUnit.MILLISECONDS);
 		then(builder).should().readTimeout(2000, TimeUnit.MILLISECONDS);
@@ -100,10 +115,22 @@ class PulsarPropertiesMapperTests {
 		properties.getAdmin().getAuthentication().setPluginClassName("myclass");
 		properties.getAdmin().getAuthentication().setParam(params);
 		PulsarAdminBuilder builder = mock(PulsarAdminBuilder.class);
-		new PulsarPropertiesMapper(properties).customizeAdminBuilder(builder);
+		new PulsarPropertiesMapper(properties).customizeAdminBuilder(builder,
+				new PropertiesPulsarConnectionDetails(properties));
 		then(builder).should().authentication("myclass", params);
 	}
 
+	@Test
+	void customizeAdminBuilderWhenHasConnectionDetails() {
+		PulsarProperties properties = new PulsarProperties();
+		properties.getAdmin().setServiceUrl("https://ignored.example.com");
+		PulsarAdminBuilder builder = mock(PulsarAdminBuilder.class);
+		PulsarConnectionDetails connectionDetails = mock(PulsarConnectionDetails.class);
+		given(connectionDetails.getAdminUrl()).willReturn("https://used.example.com");
+		new PulsarPropertiesMapper(properties).customizeAdminBuilder(builder, connectionDetails);
+		then(builder).should().serviceHttpUrl("https://used.example.com");
+	}
+
 	@Test
 	@SuppressWarnings("unchecked")
 	void customizeProducerBuilder() {

spring-boot-project/spring-boot-docker-compose/src/main/java/org/springframework/boot/docker/compose/service/connection/pulsar/PulsarDockerComposeConnectionDetailsFactory.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2012-2023 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.boot.docker.compose.service.connection.pulsar;
+
+import org.springframework.boot.autoconfigure.pulsar.PulsarConnectionDetails;
+import org.springframework.boot.docker.compose.core.ConnectionPorts;
+import org.springframework.boot.docker.compose.core.RunningService;
+import org.springframework.boot.docker.compose.service.connection.DockerComposeConnectionDetailsFactory;
+import org.springframework.boot.docker.compose.service.connection.DockerComposeConnectionSource;
+
+/**
+ * {@link DockerComposeConnectionDetailsFactory} to create {@link PulsarConnectionDetails}
+ * for a {@code pulsar} service.
+ *
+ * @author Chris Bono
+ */
+class PulsarDockerComposeConnectionDetailsFactory
+		extends DockerComposeConnectionDetailsFactory<PulsarConnectionDetails> {
+
+	private static final int BROKER_PORT = 6650;
+
+	private static final int ADMIN_PORT = 8080;
+
+	PulsarDockerComposeConnectionDetailsFactory() {
+		super("apachepulsar/pulsar");
+	}
+
+	@Override
+	protected PulsarConnectionDetails getDockerComposeConnectionDetails(DockerComposeConnectionSource source) {
+		return new PulsarDockerComposeConnectionDetails(source.getRunningService());
+	}
+
+	/**
+	 * {@link PulsarConnectionDetails} backed by a {@code pulsar} {@link RunningService}.
+	 */
+	static class PulsarDockerComposeConnectionDetails extends DockerComposeConnectionDetails
+			implements PulsarConnectionDetails {
+
+		private final String brokerUrl;
+
+		private final String adminUrl;
+
+		PulsarDockerComposeConnectionDetails(RunningService service) {
+			super(service);
+			ConnectionPorts ports = service.ports();
+			this.brokerUrl = "pulsar://%s:%s".formatted(service.host(), ports.get(BROKER_PORT));
+			this.adminUrl = "http://%s:%s".formatted(service.host(), ports.get(ADMIN_PORT));
+		}
+
+		@Override
+		public String getBrokerUrl() {
+			return this.brokerUrl;
+		}
+
+		@Override
+		public String getAdminUrl() {
+			return this.adminUrl;
+		}
+
+	}
+
+}

spring-boot-project/spring-boot-docker-compose/src/main/java/org/springframework/boot/docker/compose/service/connection/pulsar/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012-2023 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.
+ */
+
+/**
+ * Auto-configuration for docker compose Pulsar service connections.
+ */
+package org.springframework.boot.docker.compose.service.connection.pulsar;

spring-boot-project/spring-boot-docker-compose/src/main/resources/META-INF/spring.factories

@@ -19,9 +19,9 @@ org.springframework.boot.docker.compose.service.connection.oracle.OracleJdbcDock
 org.springframework.boot.docker.compose.service.connection.oracle.OracleR2dbcDockerComposeConnectionDetailsFactory,\
 org.springframework.boot.docker.compose.service.connection.postgres.PostgresJdbcDockerComposeConnectionDetailsFactory,\
 org.springframework.boot.docker.compose.service.connection.postgres.PostgresR2dbcDockerComposeConnectionDetailsFactory,\
+org.springframework.boot.docker.compose.service.connection.pulsar.PulsarDockerComposeConnectionDetailsFactory,\
 org.springframework.boot.docker.compose.service.connection.rabbit.RabbitDockerComposeConnectionDetailsFactory,\
 org.springframework.boot.docker.compose.service.connection.redis.RedisDockerComposeConnectionDetailsFactory,\
 org.springframework.boot.docker.compose.service.connection.sqlserver.SqlServerJdbcDockerComposeConnectionDetailsFactory,\
 org.springframework.boot.docker.compose.service.connection.sqlserver.SqlServerR2dbcDockerComposeConnectionDetailsFactory,\
 org.springframework.boot.docker.compose.service.connection.zipkin.ZipkinDockerComposeConnectionDetailsFactory
-

spring-boot-project/spring-boot-docker-compose/src/test/java/org/springframework/boot/docker/compose/service/connection/pulsar/PulsarDockerComposeConnectionDetailsFactoryIntegrationTests.java

@@ -0,0 +1,46 @@
+/*
+ * Copyright 2023-2023 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.boot.docker.compose.service.connection.pulsar;
+
+import org.junit.jupiter.api.Test;
+
+import org.springframework.boot.autoconfigure.pulsar.PulsarConnectionDetails;
+import org.springframework.boot.docker.compose.service.connection.test.AbstractDockerComposeIntegrationTests;
+import org.springframework.boot.testsupport.testcontainers.DockerImageNames;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Integration test for {@link PulsarDockerComposeConnectionDetailsFactory}.
+ *
+ * @author Chris Bono
+ */
+class PulsarDockerComposeConnectionDetailsFactoryIntegrationTests extends AbstractDockerComposeIntegrationTests {
+
+	PulsarDockerComposeConnectionDetailsFactoryIntegrationTests() {
+		super("pulsar-compose.yaml", DockerImageNames.pulsar());
+	}
+
+	@Test
+	void runCreatesConnectionDetails() {
+		PulsarConnectionDetails connectionDetails = run(PulsarConnectionDetails.class);
+		assertThat(connectionDetails).isNotNull();
+		assertThat(connectionDetails.getBrokerUrl()).matches("^pulsar:\\/\\/\\S+:\\d+");
+		assertThat(connectionDetails.getAdminUrl()).matches("^http:\\/\\/\\S+:\\d+");
+	}
+
+}

spring-boot-project/spring-boot-docker-compose/src/test/resources/org/springframework/boot/docker/compose/service/connection/pulsar/pulsar-compose.yaml

@@ -0,0 +1,9 @@
+services:
+  pulsar:
+    image: '{imageName}'
+    ports:
+      - '8080'
+      - '6650'
+    command: bin/pulsar standalone
+    healthcheck:
+      test: curl http://127.0.0.1:8080/admin/v2/namespaces/public/default

spring-boot-project/spring-boot-docs/src/docs/asciidoc/features/docker-compose.adoc

@@ -76,6 +76,9 @@ The following service connections are currently supported:
 | `MongoConnectionDetails`
 | Containers named "mongo"
 
+| `PulsarConnectionDetails`
+| Containers named "apachepulsar/pulsar"
+
 | `R2dbcConnectionDetails`
 | Containers named "gvenzl/oracle-xe", "mariadb", "mssql/server", "mysql", or "postgres"
 

spring-boot-project/spring-boot-docs/src/docs/asciidoc/features/testing.adoc

@@ -992,6 +992,9 @@ The following service connection factories are provided in the `spring-boot-test
 | `Neo4jConnectionDetails`
 | Containers of type `Neo4jContainer`
 
+| `PulsarConnectionDetails`
+| Containers of type `PulsarContainer`
+
 | `R2dbcConnectionDetails`
 | Containers of type `MariaDBContainer`, `MSSQLServerContainer`, `MySQLContainer`, `OracleContainer`, or `PostgreSQLContainer`
 

spring-boot-project/spring-boot-testcontainers/build.gradle

@@ -29,6 +29,7 @@ dependencies {
 	optional("org.testcontainers:neo4j")
 	optional("org.testcontainers:oracle-xe")
 	optional("org.testcontainers:postgresql")
+	optional("org.testcontainers:pulsar")
 	optional("org.testcontainers:rabbitmq")
 	optional("org.testcontainers:redpanda")
 	optional("org.testcontainers:r2dbc")
@@ -50,8 +51,8 @@ dependencies {
 	testImplementation("org.springframework:spring-r2dbc")
 	testImplementation("org.springframework.amqp:spring-rabbit")
 	testImplementation("org.springframework.kafka:spring-kafka")
+	testImplementation("org.springframework.pulsar:spring-pulsar")
 	testImplementation("org.testcontainers:junit-jupiter")
 
 	testRuntimeOnly("com.oracle.database.r2dbc:oracle-r2dbc")
 }
-

spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/pulsar/PulsarContainerConnectionDetailsFactory.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2012-2023 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.boot.testcontainers.service.connection.pulsar;
+
+import org.testcontainers.containers.PulsarContainer;
+
+import org.springframework.boot.autoconfigure.pulsar.PulsarConnectionDetails;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionDetailsFactory;
+import org.springframework.boot.testcontainers.service.connection.ContainerConnectionSource;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+
+/**
+ * {@link ContainerConnectionDetailsFactory} to create {@link PulsarConnectionDetails}
+ * from a {@link ServiceConnection @ServiceConnection}-annotated {@link PulsarContainer}.
+ *
+ * @author Chris Bono
+ */
+class PulsarContainerConnectionDetailsFactory
+		extends ContainerConnectionDetailsFactory<PulsarContainer, PulsarConnectionDetails> {
+
+	@Override
+	protected PulsarConnectionDetails getContainerConnectionDetails(ContainerConnectionSource<PulsarContainer> source) {
+		return new PulsarContainerConnectionDetails(source);
+	}
+
+	/**
+	 * {@link PulsarConnectionDetails} backed by a {@link ContainerConnectionSource}.
+	 */
+	private static final class PulsarContainerConnectionDetails extends ContainerConnectionDetails<PulsarContainer>
+			implements PulsarConnectionDetails {
+
+		private PulsarContainerConnectionDetails(ContainerConnectionSource<PulsarContainer> source) {
+			super(source);
+		}
+
+		@Override
+		public String getBrokerUrl() {
+			return getContainer().getPulsarBrokerUrl();
+		}
+
+		@Override
+		public String getAdminUrl() {
+			return getContainer().getHttpServiceUrl();
+		}
+
+	}
+
+}

spring-boot-project/spring-boot-testcontainers/src/main/java/org/springframework/boot/testcontainers/service/connection/pulsar/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright 2012-2023 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.
+ */
+
+/**
+ * Support for testcontainers Pulsar service connections.
+ */
+package org.springframework.boot.testcontainers.service.connection.pulsar;

spring-boot-project/spring-boot-testcontainers/src/main/resources/META-INF/spring.factories

@@ -19,6 +19,7 @@ org.springframework.boot.testcontainers.service.connection.kafka.KafkaContainerC
 org.springframework.boot.testcontainers.service.connection.liquibase.LiquibaseContainerConnectionDetailsFactory,\
 org.springframework.boot.testcontainers.service.connection.mongo.MongoContainerConnectionDetailsFactory,\
 org.springframework.boot.testcontainers.service.connection.neo4j.Neo4jContainerConnectionDetailsFactory,\
+org.springframework.boot.testcontainers.service.connection.pulsar.PulsarContainerConnectionDetailsFactory,\
 org.springframework.boot.testcontainers.service.connection.r2dbc.MariaDbR2dbcContainerConnectionDetailsFactory,\
 org.springframework.boot.testcontainers.service.connection.r2dbc.MySqlR2dbcContainerConnectionDetailsFactory,\
 org.springframework.boot.testcontainers.service.connection.r2dbc.OracleR2dbcContainerConnectionDetailsFactory,\

spring-boot-project/spring-boot-testcontainers/src/test/java/org/springframework/boot/testcontainers/service/connection/pulsar/PulsarContainerConnectionDetailsFactoryIntegrationTests.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2012-2023 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.boot.testcontainers.service.connection.pulsar;
+
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.containers.PulsarContainer;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
+import org.springframework.boot.autoconfigure.pulsar.PulsarAutoConfiguration;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
+import org.springframework.boot.testsupport.testcontainers.DockerImageNames;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.pulsar.annotation.PulsarListener;
+import org.springframework.pulsar.core.PulsarTemplate;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Tests for {@link PulsarContainerConnectionDetailsFactory}.
+ *
+ * @author Chris Bono
+ */
+@SpringJUnitConfig
+@Testcontainers(disabledWithoutDocker = true)
+@TestPropertySource(properties = { "spring.pulsar.consumer.subscription.initial-position=earliest" })
+class PulsarContainerConnectionDetailsFactoryIntegrationTests {
+
+	@Container
+	@ServiceConnection
+	@SuppressWarnings("unused")
+	static final PulsarContainer PULSAR = new PulsarContainer(DockerImageNames.pulsar())
+		.withStartupTimeout(Duration.ofMinutes(3));
+
+	@Autowired
+	private PulsarTemplate<String> pulsarTemplate;
+
+	@Autowired
+	private TestListener listener;
+
+	@Test
+	void connectionCanBeMadeToPulsarContainer() throws PulsarClientException {
+		this.pulsarTemplate.send("test-topic", "test-data");
+		Awaitility.waitAtMost(Duration.ofSeconds(30))
+			.untilAsserted(() -> assertThat(this.listener.messages).containsExactly("test-data"));
+	}
+
+	@Configuration(proxyBeanMethods = false)
+	@ImportAutoConfiguration(PulsarAutoConfiguration.class)
+	static class TestConfiguration {
+
+		@Bean
+		TestListener testListener() {
+			return new TestListener();
+		}
+
+	}
+
+	static class TestListener {
+
+		private final List<String> messages = new ArrayList<>();
+
+		@PulsarListener(topics = "test-topic")
+		void processMessage(String message) {
+			this.messages.add(message);
+		}
+
+	}
+
+}

spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-pulsar/build.gradle

@@ -10,6 +10,7 @@ dependencies {
 	implementation(project(":spring-boot-project:spring-boot-starters:spring-boot-starter-pulsar-reactive"))
 	testImplementation(project(":spring-boot-project:spring-boot-starters:spring-boot-starter-test"))
 	testImplementation(project(":spring-boot-project:spring-boot-tools:spring-boot-test-support"))
+	testImplementation(project(":spring-boot-project:spring-boot-testcontainers"))
 	testImplementation("org.awaitility:awaitility")
 	testImplementation("org.testcontainers:junit-jupiter")
 	testImplementation("org.testcontainers:pulsar")

spring-boot-tests/spring-boot-smoke-tests/spring-boot-smoke-test-pulsar/src/test/java/smoketest/pulsar/SamplePulsarApplicationTests.java

@@ -32,10 +32,9 @@ import org.testcontainers.junit.jupiter.Testcontainers;
 import org.springframework.boot.test.context.SpringBootTest;
 import org.springframework.boot.test.system.CapturedOutput;
 import org.springframework.boot.test.system.OutputCaptureExtension;
+import org.springframework.boot.testcontainers.service.connection.ServiceConnection;
 import org.springframework.boot.testsupport.testcontainers.DockerImageNames;
 import org.springframework.test.context.ActiveProfiles;
-import org.springframework.test.context.DynamicPropertyRegistry;
-import org.springframework.test.context.DynamicPropertySource;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -44,15 +43,10 @@ import static org.assertj.core.api.Assertions.assertThat;
 class SamplePulsarApplicationTests {
 
 	@Container
+	@ServiceConnection
 	static final PulsarContainer container = new PulsarContainer(DockerImageNames.pulsar()).withStartupAttempts(2)
 		.withStartupTimeout(Duration.ofMinutes(3));
 
-	@DynamicPropertySource
-	static void pulsarProperties(DynamicPropertyRegistry registry) {
-		registry.add("spring.pulsar.client.service-url", container::getPulsarBrokerUrl);
-		registry.add("spring.pulsar.admin.service-url", container::getHttpServiceUrl);
-	}
-
 	abstract class PulsarApplication {
 
 		private final String type;