Use Spring Boot Pulsar starters (#441)

Fixes #439

gradle.properties

@@ -10,9 +10,8 @@ pulsarClientReactiveVersion=0.3.0
 # only used by docs, tests, and samples (unpublished deps)
 springBootVersion=3.2.0-SNAPSHOT
 
-springFrameworkVersion=6.1.0-M3
+springFrameworkVersion=6.1.0-M4
 
 # these are temp until autoconfig moves into boot
-springPulsarStarterVersion=0.2.1-SNAPSHOT
 springPulsarBinderVersion=0.2.1-SNAPSHOT
 springCloudStreamVersion=4.0.3

integration-tests/integration-tests.gradle

@@ -16,10 +16,10 @@ dependencies {
 	intTestImplementation project(':spring-pulsar-test')
 	intTestRuntimeOnly 'ch.qos.logback:logback-classic'
 	intTestRuntimeOnly 'org.junit.platform:junit-platform-launcher'
-	intTestImplementation "org.springframework.boot:spring-boot-starter-test:$springBootVersion"
-	intTestImplementation "org.springframework.boot:spring-boot-starter-amqp:$springBootVersion"
-	intTestImplementation "org.springframework.pulsar:spring-pulsar-spring-boot-starter:$springPulsarStarterVersion"
-	intTestImplementation "org.springframework.pulsar:spring-pulsar-reactive-spring-boot-starter:$springPulsarStarterVersion"
+	intTestImplementation "org.springframework.boot:spring-boot-starter-test:${springBootVersion}"
+	intTestImplementation "org.springframework.boot:spring-boot-starter-amqp:${springBootVersion}"
+	intTestImplementation "org.springframework.boot:spring-boot-starter-pulsar:${springBootVersion}"
+	intTestImplementation "org.springframework.boot:spring-boot-starter-pulsar-reactive:${springBootVersion}"
 	intTestImplementation 'org.testcontainers:junit-jupiter'
 	intTestImplementation 'org.testcontainers:pulsar'
 	intTestImplementation 'org.testcontainers:rabbitmq'
@@ -41,8 +41,8 @@ task downloadRabbitConnector {
 		try {
 			download.run {
 				println "Downloading Rabbit connector to 'src/intTest/resources/connectors/' (one time only if not already downloaded)..."
-				src 'https://archive.apache.org/dist/pulsar/pulsar-2.10.2/connectors/pulsar-io-rabbitmq-2.10.2.nar'
-				dest "$buildDir/../src/intTest/resources/connectors/pulsar-io-rabbitmq-2.10.2.nar"
+				src 'https://archive.apache.org/dist/pulsar/pulsar-3.1.0/connectors/pulsar-io-rabbitmq-3.1.0.nar'
+				dest "$buildDir/../src/intTest/resources/connectors/pulsar-io-rabbitmq-3.1.0.nar"
 				overwrite false
 			}
 		} catch (Exception ex) {

integration-tests/src/intTest/java/org/springframework/pulsar/autoconfigure/PulsarFunctionAdministrationIntegrationTests.java

@@ -53,6 +53,7 @@ import org.springframework.amqp.rabbit.core.RabbitTemplate;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.WebApplicationType;
 import org.springframework.boot.autoconfigure.amqp.RabbitAutoConfiguration;
+import org.springframework.boot.autoconfigure.pulsar.PulsarAutoConfiguration;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -123,7 +124,7 @@ class PulsarFunctionAdministrationIntegrationTests {
 		app.setWebApplicationType(WebApplicationType.NONE);
 		try (ConfigurableApplicationContext context = app.run(
 				"--spring.pulsar.client.service-url=" + PULSAR_CONTAINER.getPulsarBrokerUrl(),
-				"--spring.pulsar.administration.service-url=" + PULSAR_CONTAINER.getHttpServiceUrl(),
+				"--spring.pulsar.admin.service-url=" + PULSAR_CONTAINER.getHttpServiceUrl(),
 				"--spring.rabbitmq.host=" + RABBITMQ_CONTAINER.getHost(),
 				"--spring.rabbitmq.port=" + RABBITMQ_CONTAINER.getAmqpPort())) {
 
@@ -148,7 +149,7 @@ class PulsarFunctionAdministrationIntegrationTests {
 		app.setWebApplicationType(WebApplicationType.NONE);
 		try (ConfigurableApplicationContext ignored = app.run(
 				"--spring.pulsar.client.service-url=" + PULSAR_CONTAINER.getPulsarBrokerUrl(),
-				"--spring.pulsar.administration.service-url=" + PULSAR_CONTAINER.getHttpServiceUrl(),
+				"--spring.pulsar.admin.service-url=" + PULSAR_CONTAINER.getHttpServiceUrl(),
 				"--spring.rabbitmq.host=" + RABBITMQ_CONTAINER.getHost(),
 				"--spring.rabbitmq.port=" + RABBITMQ_CONTAINER.getAmqpPort())) {
 

spring-pulsar-dependencies/spring-pulsar-dependencies.gradle

@@ -8,20 +8,20 @@ javaPlatform {
 
 dependencies {
 	api platform("com.fasterxml.jackson:jackson-bom:2.15.2")
-	api platform("io.micrometer:micrometer-bom:1.12.0-M1")
-	api platform("io.micrometer:micrometer-tracing-bom:1.2.0-M1")
-	api platform("io.projectreactor:reactor-bom:2023.0.0-M1")
+	api platform("io.micrometer:micrometer-bom:1.12.0-M2")
+	api platform("io.micrometer:micrometer-tracing-bom:1.2.0-M2")
+	api platform("io.projectreactor:reactor-bom:2023.0.0-M2")
 	api platform("io.zipkin.brave:brave-bom:5.16.0")
 	api platform("org.assertj:assertj-bom:3.24.2")
 	api platform("org.awaitility:awaitility:4.2.0")
-	api platform("org.junit:junit-bom:5.9.3")
-	api platform("org.mockito:mockito-bom:5.4.0")
+	api platform("org.junit:junit-bom:5.10.0")
+	api platform("org.mockito:mockito-bom:5.5.0")
 	api platform("org.springframework:spring-framework-bom:$springFrameworkVersion")
-	api platform("org.testcontainers:testcontainers-bom:1.18.3")
+	api platform("org.testcontainers:testcontainers-bom:1.19.0")
 
 	constraints {
-		api "ch.qos.logback:logback-classic:1.4.8"
-		api "com.github.ben-manes.caffeine:caffeine:3.1.6"
+		api "ch.qos.logback:logback-classic:1.4.11"
+		api "com.github.ben-manes.caffeine:caffeine:3.1.8"
 		api "com.google.code.findbugs:jsr305:3.0.2"
 		api "com.google.protobuf:protobuf-java:3.21.5"
 		api "com.jayway.jsonpath:json-path:2.8.0"

spring-pulsar-docs/spring-pulsar-docs.gradle

@@ -90,7 +90,6 @@ tasks.withType(org.asciidoctor.gradle.jvm.AbstractAsciidoctorTask) {
 				"spring-cloud-stream-version": project.springCloudStreamVersion ?: 'current',
 				"spring-pulsar-version": project.version,
 				"spring-pulsar-binder-version": project.springPulsarBinderVersion ?: 'current',
-				"spring-pulsar-starter-version": project.springPulsarStarterVersion ?: 'current',
 				"pulsar-client-version": project.pulsarClientVersion ?: 'current',
 				"pulsar-client-reactive-version": project.pulsarClientReactiveVersion ?: 'current',
 				"is-snapshot-version": project.version.endsWith("-SNAPSHOT")

spring-pulsar-docs/src/main/asciidoc/attributes-variables.adoc

@@ -3,7 +3,6 @@
 :spring-cloud-stream-version: current
 :spring-pulsar-version: current
 :spring-pulsar-binder-version: current
-:spring-pulsar-starter-version: current
 :pulsar-client-version: current
 :pulsar-client-reactive-version: current
 :is-snapshot-version:  false

spring-pulsar-docs/src/main/asciidoc/pulsar-admin.adoc

@@ -10,15 +10,15 @@ The administration implements an interface called `PulsarAdminOperations` and pr
 When you use the Pulsar Spring Boot starter, you get the `PulsarAdministration` auto-configured.
 
 By default, the application tries to connect to a local Pulsar instance at `\http://localhost:8080`.
-This can be adjusted by setting the `spring.pulsar.administration.service-url` property to a different value in the form `(http|https)://<host>:<port>`.
+This can be adjusted by setting the `spring.pulsar.admin.service-url` property to a different value in the form `(http|https)://<host>:<port>`.
 
 There are many application properties available to configure the client.
-See the {spring-boot-pulsar-config-props}[`spring.pulsar.administration.*`] application properties.
+See the {spring-boot-pulsar-config-props}[`spring.pulsar.admin.*`] application properties.
 
 [[pulsar-admin-authentication]]
 === Authentication
 When accessing a Pulsar cluster that requires authentication, the admin client requires the same security configuration as the regular Pulsar client.
-You can use the aforementioned <<pulsar.adoc#client-authentication,security configuration>> by replacing `spring.pulsar.client` with `spring.pulsar.administration`.
+You can use the aforementioned <<pulsar.adoc#client-authentication,security configuration>> by replacing `spring.pulsar.client` with `spring.pulsar.admin`.
 
 [[pulsar-auto-topic-creation]]
 == Automatic Topic Creation

spring-pulsar-docs/src/main/asciidoc/pulsar-binder.adoc

@@ -250,7 +250,7 @@ Spring Cloud Stream binder for Apache Pulsar comes with an out-of-the-box provis
 When running an application, if the necessary topics are absent, Pulsar will create the topics for you.
 However, this is a basic non-partitioned topic, and if you want advanced features like creating a partitioned topic, you can rely on the topic provisioner in the binder.
 Pulsar topic provisioner uses `PulsarAdministration` from the framework, which uses the `PulsarAdminBuilder.`
-For this reason, you need to set the `spring.pulsar.administration.service-url` property unless you are running Pulsar on the default server and port.
+For this reason, you need to set the `spring.pulsar.admin.service-url` property unless you are running Pulsar on the default server and port.
 
 === Specifying partition count when creating the topic
 

spring-pulsar-docs/src/main/asciidoc/quick-tour.adoc

@@ -13,9 +13,9 @@ Spring Boot applications need only the `spring-pulsar-spring-boot-starter` depen
 ----
 <dependencies>
     <dependency>
-        <groupId>org.springframework.pulsar</groupId>
-        <artifactId>spring-pulsar-spring-boot-starter</artifactId>
-        <version>{spring-pulsar-starter-version}</version>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-pulsar</artifactId>
+        <version>{spring-boot-version}</version>
     </dependency>
 </dependencies>
 ----
@@ -24,7 +24,7 @@ Spring Boot applications need only the `spring-pulsar-spring-boot-starter` depen
 .Gradle
 ----
 dependencies {
-    implementation 'org.springframework.pulsar:spring-pulsar-spring-boot-starter:{spring-pulsar-starter-version}'
+    implementation 'org.springframework.boot:spring-boot-starter-pulsar:{spring-boot-version}'
 }
 ----
 

spring-pulsar-docs/src/main/asciidoc/reactive-quick-tour.adoc

@@ -14,9 +14,9 @@ Spring Boot applications need only the `spring-pulsar-reactive-spring-boot-start
 ----
 <dependencies>
     <dependency>
-        <groupId>org.springframework.pulsar</groupId>
-        <artifactId>spring-pulsar-reactive-spring-boot-starter</artifactId>
-        <version>{spring-pulsar-starter-version}</version>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-pulsar-reactive</artifactId>
+        <version>{spring-boot-version}</version>
     </dependency>
 </dependencies>
 ----
@@ -24,7 +24,7 @@ Spring Boot applications need only the `spring-pulsar-reactive-spring-boot-start
 .Gradle
 ----
 dependencies {
-    implementation 'org.springframework.pulsar:spring-pulsar-reactive-spring-boot-starter:{spring-pulsar-starter-version}'
+    implementation 'org.springframework.boot:spring-boot-starter-pulsar-reactive:{spring-boot-version}'
 }
 ----
 

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/listener/ReactivePulsarListenerTests.java

@@ -185,11 +185,14 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 			assertThat(containerProperties.getTopicsPattern().toString())
 				.isEqualTo("persistent://public/default/pattern.*");
 
+			// Let things setup before firing the messages
+			Thread.sleep(2000);
+
 			pulsarTemplate.send("persistent://public/default/pattern-1", "hello baz");
 			pulsarTemplate.send("persistent://public/default/pattern-2", "hello baz");
 			pulsarTemplate.send("persistent://public/default/pattern-3", "hello baz");
 
-			assertThat(latch3.await(10, TimeUnit.SECONDS)).isTrue();
+			assertThat(latch3.await(15, TimeUnit.SECONDS)).isTrue();
 		}
 
 		@EnableReactivePulsar
@@ -197,12 +200,17 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 		static class TestPulsarListenersForBasicScenario {
 
 			@ReactivePulsarListener(id = "id-1", topics = "topic-1", subscriptionName = "subscription-1",
-					consumerCustomizer = "consumerCustomizer")
+					consumerCustomizer = "listen1Customizer")
 			Mono<Void> listen1(String ignored) {
 				latch1.countDown();
 				return Mono.empty();
 			}
 
+			@Bean
+			ReactiveMessageConsumerBuilderCustomizer<String> listen1Customizer() {
+				return b -> b.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
+			}
+
 			@ReactivePulsarListener(consumerCustomizer = "listen2Customizer")
 			Mono<Void> listen2(String ignored) {
 				latch2.countDown();
@@ -216,15 +224,15 @@ public class ReactivePulsarListenerTests implements PulsarTestContainerSupport {
 			}
 
 			@ReactivePulsarListener(id = "id-3", topicPattern = "persistent://public/default/pattern.*",
-					subscriptionName = "subscription-3", consumerCustomizer = "consumerCustomizer")
+					subscriptionName = "subscription-3", consumerCustomizer = "listen3Customizer")
 			Mono<Void> listen3(String ignored) {
 				latch3.countDown();
 				return Mono.empty();
 			}
 
 			@Bean
-			ReactiveMessageConsumerBuilderCustomizer<String> consumerCustomizer() {
-				return b -> b.topicsPatternAutoDiscoveryPeriod(Duration.ofSeconds(2))
+			ReactiveMessageConsumerBuilderCustomizer<String> listen3Customizer() {
+				return b -> b.topicsPatternAutoDiscoveryPeriod(Duration.ofSeconds(5))
 					.subscriptionInitialPosition(SubscriptionInitialPosition.Earliest);
 			}
 

spring-pulsar-sample-apps/sample-app1/build.gradle

@@ -6,14 +6,14 @@ plugins {
 
 description = 'Spring Pulsar Sample Application (Send and Receive)'
 
-repositories {
+repositories { 
 	mavenCentral()
 	maven { url 'https://repo.spring.io/milestone' }
 	maven { url 'https://repo.spring.io/snapshot' }
 }
 
 dependencies {
-	implementation "org.springframework.pulsar:spring-pulsar-spring-boot-starter:${springPulsarStarterVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-pulsar:${springBootVersion}"
 	implementation 'org.springframework.boot:spring-boot-starter-actuator'
 	implementation 'io.micrometer:micrometer-tracing-bridge-brave'
 	implementation 'io.zipkin.reporter2:zipkin-reporter-brave'

spring-pulsar-sample-apps/sample-app2/build.gradle

@@ -6,14 +6,14 @@ plugins {
 
 description = 'Spring Pulsar Sample Applications (Custom Routing)'
 
-repositories {
+repositories { 
 	mavenCentral()
 	maven { url 'https://repo.spring.io/milestone' }
 	maven { url 'https://repo.spring.io/snapshot' }
 }
 
 dependencies {
-	implementation "org.springframework.pulsar:spring-pulsar-spring-boot-starter:${springPulsarStarterVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-pulsar:${springBootVersion}"
 	implementation 'org.springframework.boot:spring-boot-starter-actuator'
 	implementation 'io.micrometer:micrometer-tracing-bridge-brave'
 	implementation 'io.zipkin.reporter2:zipkin-reporter-brave'

spring-pulsar-sample-apps/sample-pulsar-binder/build.gradle

@@ -6,15 +6,15 @@ plugins {
 
 description = 'Spring Cloud Stream Binder for Pulsar Sample Application'
 
-repositories {
+repositories { 
 	mavenCentral()
 	maven { url 'https://repo.spring.io/milestone' }
 	maven { url 'https://repo.spring.io/snapshot' }
 }
 
 dependencies {
+	implementation "org.springframework.boot:spring-boot-starter-pulsar:${springBootVersion}"
 	implementation "org.springframework.pulsar:spring-pulsar-spring-cloud-stream-binder:${springPulsarBinderVersion}"
-	implementation "org.springframework.pulsar:spring-pulsar-spring-boot-starter:${springPulsarStarterVersion}"
 }
 
 bootRun {

spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/build.gradle

@@ -7,7 +7,7 @@ plugins {
 group = 'org.springframework.pulsar.sample'
 description = 'Sample Signup App (Pulsar Functions)'
 
-repositories {
+repositories { 
 	mavenCentral()
 	maven { url 'https://repo.spring.io/milestone' }
 	maven { url 'https://repo.spring.io/snapshot' }
@@ -15,7 +15,7 @@ repositories {
 
 dependencies {
 	implementation "org.springframework.pulsar:spring-pulsar-spring-cloud-stream-binder:${springPulsarBinderVersion}"
-	implementation "org.springframework.pulsar:spring-pulsar-spring-boot-starter:${springPulsarStarterVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-pulsar:${springBootVersion}"
 	implementation 'org.springframework.boot:spring-boot-starter-amqp'
 	implementation 'org.springframework.boot:spring-boot-starter-data-cassandra'
 	implementation 'org.springframework.boot:spring-boot-starter-actuator'

spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-function/build.gradle

@@ -6,7 +6,7 @@ group = 'org.springframework.pulsar.sample'
 description = 'Sample Signup Pulsar Function'
 sourceCompatibility = '1.8'
 
-repositories {
+repositories { 
 	mavenCentral()
 }
 

spring-pulsar-sample-apps/sample-pulsar-reader/build.gradle

@@ -6,14 +6,14 @@ plugins {
 
 description = 'Spring Pulsar Sample Application (Reader)'
 
-repositories {
+repositories { 
 	mavenCentral()
 	maven { url 'https://repo.spring.io/milestone' }
 	maven { url 'https://repo.spring.io/snapshot' }
 }
 
 dependencies {
-	implementation "org.springframework.pulsar:spring-pulsar-spring-boot-starter:${springPulsarStarterVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-pulsar:${springBootVersion}"
 }
 
 bootRun {

spring-pulsar-sample-apps/sample-reactive/build.gradle

@@ -1,19 +1,19 @@
 plugins {
 	id 'java'
-	id 'org.springframework.boot' version '3.0.8-SNAPSHOT'
+	id 'org.springframework.boot' version '3.2.0-SNAPSHOT'
 	id 'io.spring.dependency-management' version '1.1.0'
 }
 
 description = 'Reactive Spring Pulsar Sample Application'
 
-repositories {
+repositories { 
 	mavenCentral()
 	maven { url 'https://repo.spring.io/milestone' }
 	maven { url 'https://repo.spring.io/snapshot' }
 }
 
 dependencies {
-	implementation "org.springframework.pulsar:spring-pulsar-reactive-spring-boot-starter:${springPulsarStarterVersion}"
+	implementation "org.springframework.boot:spring-boot-starter-pulsar-reactive:${springBootVersion}"
 }
 
 bootRun {