diff --git a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/asciidoc/AsciidoctorConventionsPlugin.java b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/asciidoc/AsciidoctorConventionsPlugin.java index f1f75ec1..5dbd9c78 100644 --- a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/asciidoc/AsciidoctorConventionsPlugin.java +++ b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/asciidoc/AsciidoctorConventionsPlugin.java @@ -133,6 +133,7 @@ public class AsciidoctorConventionsPlugin implements Plugin { attributes.put("attribute-missing", "warn"); attributes.put("github-tag", determineGitHubTag(project)); attributes.put("revnumber", null); + attributes.put("allow-uri-read", ""); // allow remote includes asciidoctorTask.attributes(attributes); } diff --git a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/ConfigurationPropertiesPlugin.java b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/ConfigurationPropertiesPlugin.java index d0bc788d..84fd067c 100644 --- a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/ConfigurationPropertiesPlugin.java +++ b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/ConfigurationPropertiesPlugin.java @@ -79,11 +79,6 @@ public class ConfigurationPropertiesPlugin implements Plugin { private void addConfigurationProcessorDependency(Project project) { Configuration annotationProcessors = project.getConfigurations() .getByName(JavaPlugin.ANNOTATION_PROCESSOR_CONFIGURATION_NAME); -// Object bootVersion = project.findProperty("springBootVersion"); -// if (bootVersion == null) { -// bootVersion = project.findProperty("spring-boot.version"); -// } -// Assert.notNull(bootVersion, "Unable to determine Spring Boot version"); annotationProcessors.getDependencies().add(project.getDependencies().create( "org.springframework.boot:spring-boot-configuration-processor")); } diff --git a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java index 5e73f004..5bcb2dbd 100644 --- a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java +++ b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java @@ -63,12 +63,12 @@ public class DocumentConfigurationProperties extends DefaultTask { @TaskAction void documentConfigurationProperties() throws IOException { Snippets snippets = new Snippets(this.configurationPropertyMetadata); - snippets.add("application-properties.pulsar", "Pulsar Properties", this::pulsarPrefixes); + snippets.add("application-properties.pulsar-client", "Pulsar Client Properties", (c) -> c.accept("spring.pulsar.client")); + snippets.add("application-properties.pulsar-producer", "Pulsar Producer Properties", (c) -> c.accept("spring.pulsar.producer")); + snippets.add("application-properties.pulsar-consumer", "Pulsar Consumer Properties", (c) -> { + c.accept("spring.pulsar.consumer"); + c.accept("spring.pulsar.listener"); + }); snippets.writeTo(this.outputDir.toPath()); } - - private void pulsarPrefixes(Snippet.Config config) { - config.accept("spring.pulsar"); - } - } diff --git a/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc b/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc index 04ac80df..0f94d9d9 100644 --- a/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc @@ -10,4 +10,8 @@ TIP: Spring Boot provides various conversion mechanism with advanced value forma NOTE: Property contributions can come from additional jar files on your classpath, so you should not consider this an exhaustive list. Also, you can define your own properties. -include::application-properties/pulsar.adoc[] +include::application-properties/pulsar-client.adoc[] + +include::application-properties/pulsar-producer.adoc[] + +include::application-properties/pulsar-consumer.adoc[] diff --git a/spring-pulsar-docs/src/main/asciidoc/index.adoc b/spring-pulsar-docs/src/main/asciidoc/index.adoc index 9538eac2..f8e0e490 100644 --- a/spring-pulsar-docs/src/main/asciidoc/index.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/index.adoc @@ -5,7 +5,7 @@ :numbered: :icons: font :hide-uri-scheme: -Soby Chacko +Soby Chacko; Chris Bono //ifdef::backend-html5[] //*{project-version}* diff --git a/spring-pulsar-docs/src/main/asciidoc/preface.adoc b/spring-pulsar-docs/src/main/asciidoc/preface.adoc index 706850a8..b9326bc9 100644 --- a/spring-pulsar-docs/src/main/asciidoc/preface.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/preface.adoc @@ -1,3 +1,4 @@ -This project provides a basic Spring friendly API for developing https://pulsar.apache.org/[Apache Pulsar] applications. +This project provides a basic Spring-friendly API for developing https://pulsar.apache.org/[Apache Pulsar] applications. + On a very high-level, Spring for Apache Pulsar provides a `PulsarTemplate` for publishing to a Pulsar topic and a `PulsarListener` annotation for consuming from it. -In addition, it also provides various convenient APIs for Sprind developers on ramp their development journey into Aapche Pulsar. +In addition, it also provides various convenient APIs for Spring developers to ramp up their development journey into Aapche Pulsar. diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc index c58e0aee..3c70d104 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar.adoc @@ -1,5 +1,7 @@ [[pulsar]] == Using Spring for Apache Pulsar +:javadocs: https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/api +:github: https://github.com/spring-projects-experimental/spring-pulsar This section offers detailed explanations of the various concerns that impact using Spring for Apache Pulsar. For a quick but less detailed introduction, see <>. @@ -7,102 +9,17 @@ For a quick but less detailed introduction, see <>. [[pulsar-client]] === Pulsar Client -When using `spring-pulsar-spring-boot-autoconfigure`, you get the `PulsarClient` auto-configured. +When using the Pulsar Spring Boot Starter, you get the `PulsarClient` auto-configured. This is done through a factory bean called `PulsarClientFactoryBean`, which takes a configuration object `PulsarClientConfiguration`. -By default, the application tries to connect a local Pulsar instance available at `pulsar://localhost:6650`. -The properties listed below are available to be set on the Pulsar client. - -The same defaults expected by the Pulsar client are honored when no configuration provided for these properties. - -``` -spring.pulsar.client.serviceUrl -spring.pulsar.client.authPluginClassName -spring.pulsar.client.authParams -spring.pulsar.client.operationTimeoutMs -spring.pulsar.client.statsIntervalSeconds -spring.pulsar.client.numIoThreads -spring.pulsar.client.useTcpNoDelayuseTls -spring.pulsar.client.tlsAllowInsecureConnection -spring.pulsar.client.tlsHostnameVerificationEnable -spring.pulsar.client.concurrentLookupRequest -spring.pulsar.client.maxLookupRequest -spring.pulsar.client.maxNumberOfRejectedRequestPerConnection -spring.pulsar.client.keepAliveIntervalSeconds -spring.pulsar.client.connectionTimeoutMs -spring.pulsar.client.requestTimeoutMs -spring.pulsar.client.initialBackoffIntervalNanos -spring.pulsar.client.maxBackoffIntervalNanos; -``` +NOTE: By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`. However, there are many <> available to configure the client. When no configuration is provided for a property it's default value will be used. [[pulsar-producer]] === Pulsar Producer -On the Pulsar producer side, Spring Boot auto-configuration will provide a `PulsarTemplate` which is backed by a `PulsarProducerFactory`. -We will see more details of these components later in this document, but in this section, let us look at the configuration properties available on the producer. +On the Pulsar producer side, Spring Boot auto-configuration provides a `PulsarTemplate` for publishing records. The template implements an interface called `PulsarOperations` and provides {javadocs}/org/springframework/pulsar/core/PulsarOperations.html[several variants of 'send' methods] to publish records through its contract. -The same defaults expected by the Pulsar producer are honored when no configuration provided for these properties. - -``` -spring.pulsar.producer.topicName -spring.pulsar.producer.producerName -spring.pulsar.producer.sendTimeoutMs -spring.pulsar.producer.blockIfQueueFull -spring.pulsar.producer.maxPendingMessages -spring.pulsar.producer.maxPendingMessagesAcrossPartitions -spring.pulsar.producer.messageRoutingMode -spring.pulsar.producer.hashingScheme -spring.pulsar.producer.cryptoFailureAction -spring.pulsar.producer.batchingMaxPublishDelayMicros -spring.pulsar.producer.batchingMaxMessages -spring.pulsar.producer.batchingEnabled -spring.pulsar.producer.chunkingEnabled -spring.pulsar.producer.compressionType -spring.pulsar.producer.initialSubscriptionName -spring.pulsar.producer.accessMode -``` - -`PulsarTemplate` implements an interface called `PulsarOperations` and provides the following variants of the `send` method to publish records through its contract. - -==== -[source, java] ----- -MessageId send(T message) throws PulsarClientException; - -MessageId send(String topic, T message) throws PulsarClientException; - -MessageId send(T message, MessageRouter messageRouter) throws PulsarClientException; - -MessageId send(T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer) throws PulsarClientException; - -MessageId send(String topic, T message, MessageRouter messageRouter) throws PulsarClientException; - -MessageId send(T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException; - -MessageId send(String topic, T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer) throws PulsarClientException; - -MessageId send(String topic, T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException; - -CompletableFuture sendAsync(T message) throws PulsarClientException; - -CompletableFuture sendAsync(String topic, T message) throws PulsarClientException; - -CompletableFuture sendAsync(T message, MessageRouter messageRouter) throws PulsarClientException; - -CompletableFuture sendAsync(T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer) throws PulsarClientException; - -CompletableFuture sendAsync(String topic, T message, MessageRouter messageRouter) throws PulsarClientException; - -CompletableFuture sendAsync(T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException; - -CompletableFuture sendAsync(String topic, T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer) throws PulsarClientException; - -CompletableFuture sendAsync(String topic, T message, TypedMessageBuilderCustomizer typedMessageBuilderCustomizer, MessageRouter messageRouter) throws PulsarClientException; - ----- -==== - -There are two categories of send API methods exposed by `PulsarTemplate` - `send` and `sendAsync`. +There are two categories of these send API methods - `send` and `sendAsync`. The `send` methods are blocking calls using the synchronous sending capabilities on the Pulsar producer. They return the `MessageId` of the message that was published once the message is persisted on the broker. The `sendAsync` method calls are asynchronous calls that are non-blocking. @@ -124,69 +41,29 @@ template.send(msg, (messageBuilder -> messageBuilder.key(myMessageKey))); When using partitioned topics, then you can use custom partitioning routing when publishing records. For this purpose, you can provide an implementation of a `MessageRouter` and pass it along with both `send` and `sendAsync` methods. -Note that, when using a `MessageRouter, you must set the `spring.pulsar.producer.messageRoutingMode` property to `custom`. +Note that, when using a `MessageRouter`, you must set the `spring.pulsar.producer.messageRoutingMode` property to `custom`. + +NOTE: There are many <> available to configure the producer. + [[pulsar-producer-factory]] === Pulsar Producer Factory - -`PulsarTemplate` uses a `PulsarProducerFactory` for creating the underlying Pulsar producer. -When using Spring Boot through `spring-pulsar-spring-boot-autoconfigure`, then it automatically autoconfigures a `PulsarProducerFactory`. -Any producer properties mentioned above (using the prefix, `spring.pulsar.producer`) is passed along to the backing producer factory implementation along with a Pulsar client. -You can disable the autoconfigured producer factory by providing your own bean definition for `DefaultPulsarProducerFactory` in the application. - -`PulsarProducerFactory` provides the following API methods to create a producer. - -==== -[source, java] ----- -Producer createProducer(String topic, Schema schema) throws PulsarClientException; - -Producer createProducer(String topic, Schema schema, MessageRouter messageRouter) throws PulsarClientException; ----- -==== +The `PulsarTemplate` relies on a `PulsarProducerFactory` for actually creating the underlying producer. Spring Boot auto-configuration also provides this producer factory. Additionally, you can configure the factory by specifying any of the available producer-centric <>. [[producer-caching]] === Pulsar Producer Caching - -//TODO +**TODO** [[pulsar-listener]] === Pulsar Listener When it comes to Pulsar consumer, we recommend the end user applications to make use of the `PulsarListener` annotation. -In order to use `PulsarListener`, you need to use the `EnablePulsar` annotation. +In order to use `PulsarListener`, you need to use the `@EnablePulsar` annotation. When using the Spring Boot support, it automatically enables this annotation and configures all the components necessary for `PulsarListener` such as the message listener infrastructure which is responsible for creating the Pulsar consumer. `PulsarMessageListenerContainer` uses a `PulsarConsumerFactory` in order to create and manage the Pulsar consumer. -This consumer factory is auto-configured through Spring Boot. -Following are the consumer properties that you can configure through the Boot support. +This consumer factory is also auto-configured through Spring Boot. -The same defaults expected by the Pulsar consumer are honored when no configuration provided for these properties. - -``` -spring.pulsar.consumer.topicNames -spring.pulsar.consumer.topicsPattern -spring.pulsar.consumer.subscriptionName -spring.pulsar.consumer.subscriptionType -spring.pulsar.consumer.receiverQueueSize -spring.pulsar.consumer.acknowledgementsGroupTimeMicros -spring.pulsar.consumer.negativeAckRedeliveryDelayMicros -spring.pulsar.consumer.maxTotalReceiverQueueSizeAcrossPartitions -spring.pulsar.consumer.consumerName -spring.pulsar.consumer.ackTimeoutMillis -spring.pulsar.consumer.tickDurationMillis -spring.pulsar.consumer.priorityLevel -spring.pulsar.consumer.cryptoFailureAction -spring.pulsar.consumer.properties -spring.pulsar.consumer.readCompacted -spring.pulsar.consumer.subscriptionInitialPosition -spring.pulsar.consumer.patternAutoDiscoveryPeriod -spring.pulsar.consumer.regexSubscriptionMode -spring.pulsar.consumer.autoUpdatePartitions -spring.pulsar.consumer.replicateSubscriptionState -spring.pulsar.consumer.autoAckOldestChunkedMessageOnQueueFull -spring.pulsar.consumer.maxPendingChunkedMessageexpireTimeOfIncompleteChunkedMessageMillis -spring.pulsar.consumer.maxPendingChunkedMessageexpireTimeOfIncompleteChunkedMessageMillis -``` +NOTE: There are many consumer-centric <> available to configure these components. Let us revisit the `PulsarListener` code snippet we saw in the quick-tour section. @@ -214,10 +91,12 @@ public void listen(String message) { In this most basic form, you must provide the following two properties with their corresponding values. -``` -spring.pulsar.consumer.topicNames=hello-pulsar -spring.pulsar.consumer.subscriptionName=hello-pulsar-subscription -``` +[source,yaml,indent=0,subs="verbatim"] +---- +spring.pulsar.consumer: + topic-names: hello-pulsar + subscription-name: hello-pulsar-subscription +---- In the `PulsarListener` method above, we receive the data as `String`, but we don't specify any schema types. Internally, the framework relies on Pulsar's schema mechanism to convert the data to the required type. @@ -261,9 +140,7 @@ Here is an example of using `PulsarListener` to consume records in batches. @PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) public void listen4(List messages) { System.out.println("records received :" + messages.size()); - for (Foo message : messages) { - System.out.println("record : " + message); - } + messages.forEach((message) -> System.out.println("record : " + message)); } ---- ==== @@ -282,9 +159,7 @@ The following also should work in which we use the `Messages` holder type provid @PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) public void listen4(Messages messages) { System.out.println("records received :" + messages.size()); - for (Foo message : messages) { - System.out.println("record : " + message); - } + messages.forEach((message) -> System.out.println("record : " + message)); } ---- ==== @@ -305,26 +180,13 @@ Once data is received, it is handed over to the selected message listener implem The following message listener types are available when using Spring for Apache Pulsar. -==== -[source, java] ----- -interface PulsarRecordMessageListener extends MessageListener{ - void received(Consumer consumer, Message msg); -} +* link:{github}/blob/main/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarRecordMessageListener.java#L29[PulsarRecordMessageListener] -interface PulsarAcknowledgingMessageListener extends PulsarRecordMessageListener { - void received(Consumer consumer, Message msg, Acknowledgement acknowledgement); -} +* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarAcknowledgingMessageListener.java#L28[PulsarAcknowledgingMessageListener] -interface PulsarBatchMessageListener extends PulsarRecordMessageListener { - void received(Consumer consumer, Messages msg); -} +* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchMessageListener.java#L36[PulsarBatchMessageListener] -interface PulsarBatchAcknowledgingMessageListener extends PulsarBatchMessageListener { - void received(Consumer consumer, Messages msg, Acknowledgement acknowledgement); -} ----- -==== +* link:{github}/blob/ade2c74482d8ac1407ffe4840fa058475c07bcfc/spring-pulsar/src/main/java/org/springframework/pulsar/listener/PulsarBatchAcknowledgingMessageListener.java#L28[PulsarBatchAcknowledgingMessageListener] We will see the details about these various message listeners in the sections below. @@ -362,9 +224,7 @@ Here is the `PulsarListener` example of consuming records in batches. @PulsarListener(subscriptionName = "hello-batch-subscription", topics = "hello-batch", schemaType = SchemaType.JSON, batch = true) public void listen4(List messages) { System.out.println("records received :" + messages.size()); - for (Foo message : messages) { - System.out.println("record : " + message); - } + messages.forEach((message) -> System.out.println("record : " + message)); } ---- ==== @@ -455,7 +315,9 @@ You can also negatively acknowledge with the `MessageId` for the batch listener. In the sample below, we are publishing to a topic called `hello-pulsar-partitioned`. It is a topic that is partitioned and for this sample we assume that the topic is already created with three partitions. -``` +==== +[source, java] +---- @SpringBootApplication public class PulsarBootPartitioned { @@ -505,7 +367,8 @@ public class PulsarBootPartitioned { } } -``` +---- +==== A few things require explanation in the application above. We are publishing to a partitioned topic and we would like to publish some data segment to a specific partition. @@ -524,7 +387,9 @@ We can switch to the `failover` subscription mode and add three separate consume Here is an example. -``` +==== +[source, java] +---- @PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "failover") public void listen1(String foo) { System.out.println("Message Received 1: " + foo); @@ -536,10 +401,11 @@ public void listen2(String foo) { } @PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "failover") -public void liste3n(String foo) { +public void listen3(String foo) { System.out.println("Message Received 3: " + foo); } -``` +---- +==== When following this approach, you can see that a single partition always gets consumed by a dedicated consumer. @@ -548,7 +414,9 @@ Keep in mind though, that when using the `shared` mode, you lose any ordering gu Here is an example. -``` +==== +[source, java] +---- @PulsarListener(subscriptionName = "hello-pulsar-shared-subscription", topics = "hello-pulsar-partitioned", subscriptionType = "shared") public void listen1(String foo) { System.out.println("Message Received 1: " + foo); @@ -558,46 +426,56 @@ public void listen1(String foo) { public void listen2(String foo) { System.out.println("Message Received 2: " + foo); } -``` +---- +==== === Accessing the Pulsar Message Object In your `PulsarListener` method, you can receive the record directly as a Pulsar Message instead of the actual payload type. Here is an example. -``` +==== +[source, java] +---- @PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") public void listen(org.apache.pulsar.client.api.Message message) { System.out.println("Data Received: " + message.getValue()); } -``` +---- +==== === Accessing the Pulsar Messages Object When consuming messages in batch mode using `PulsarListener`, instead of receiving them as a `List, you can receive them as Pulsar Messages type. Here is an example. -``` +==== +[source, java] +---- @PulsarListener(subscriptionName = "batch-subscription", topics = "hello-pulsar", batch = "true") public void listen(org.apache.pulsar.client.api.Messages messages) { // Iterate on the messages // Each iteration gives access to a org.apache.pulsar.client.api.Message object } -``` +---- +==== === Accessing the Pulsar Consumer Object Sometimes, it is necessary to gain direct access to the Pulsar Consumer object. Here is how you may do so. -``` +==== +[source, java] +---- @PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") public void listen(String message, org.apache.pulsar.client.api.Consumer consumer) { System.out.println("Message Received: " + message); ConsumerStats stats = consumer.getStats(); ... } -``` +---- +==== When accessing the `Consumer` object this way, make sure NOT to invoke any operations that would change the Consumer's cursor position by invoking any receive methods. All such operations must be done by the container. @@ -608,12 +486,15 @@ As indicated above, for normal Java types (the primitive ones), Spring Pulsar fr However, for more complex types such as JSON or AVRO, you need to specify the schema type on the annotation. Here is how you provide that. -``` +==== +[source, java] +---- @PulsarListener(subscriptionName = "json-subscription", topics = "hello-pulsar-json", schemaType = SchemaType.JSON) public void listen(Foo foo) { System.out.println("Message received: " + foo); } -``` +---- +==== On the producer side also, for the Java primitive types, the framework can infer the Schema, but for any other types, you need set that on the `PulsarTemmplate`. diff --git a/spring-pulsar-docs/src/main/asciidoc/quick-tour.adoc b/spring-pulsar-docs/src/main/asciidoc/quick-tour.adoc index 331757b9..8695db82 100644 --- a/spring-pulsar-docs/src/main/asciidoc/quick-tour.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/quick-tour.adoc @@ -3,86 +3,68 @@ In this section, we will take a quick tour of Spring for Apache Pulsar. -#### Minimum Supported Versions +==== Minimum Supported Versions -The following are the minimum supported versions of the underlying libraries required by Spring for Apache Pulsar +The minimum supported versions for the underlying libraries required by the framework are as follows: +|=== +| Library | Version -**JDK** - JDK 17 +| Java +| 17 -**Apache Pulsar** - 2.10.0 +| Apache Pulsar +| 2.10.0 -**Spring Boot** - 3.0.0 +| Spring Boot +| 3.0.0 -**Spring Framework** - 6.0.0 +| Spring Framework +| 6.0.0 -### Building the Project +| Gradle +| 7.x (7.5 or later) +|=== -Spring for Apache Pulsar uses gradle as it's build tool. -Use the following command to do a full build of the project. +=== Building the Project -``` +Spring for Apache Pulsar uses Gradle as it's build tool. Run the following command to do a full build of the project. +[indent=0] +---- ./gradlew clean build -``` +---- +You can build without running tests by using the following command: +[indent=0] +---- +./gradlew clean build -x test +---- -The build will produce the following artifacts. - -* spring-pulsar -* spring-pulsar-spring-boot-autoconfigure - -### Maven Coordinates - -Here are the maven coordinates of the modules, if you want to directly include them in an application. - -Here are the maven coordinates of the artifacts needed for using this library. - -``` - - org.springframework.pulsar - spring-pulsar - 0.1.0-SNAPSHOT - - - - org.springframework.pulsar - spring-pulsar-boot-autoconfiguration - 0.1.0-SNAPSHOT - -``` - -If you include the `spring-pulsar-boot-autoconfiguration` module, then that brings the `spring-pulsar` module transitively. - -We recommend the usage of a Spring-Boot-First-Approach for Spring for Apache Pulsar based application as that simplifies things tremendously. -Therefore, most often, simply including the `spring-pulsar-boot-autoconfiguration` module is all you need in the application's dependencies. - -### Quick Sample +=== Quick Sample In the following sample Spring Boot application, we show how to write a publisher and consumer using Spring for Apache Pulsar. This is a complete application and does not require any additional configuration as long as you have Pulsar cluster running on the default location - `localhost:6650`. -``` +NOTE: We recommend the usage of a Spring-Boot-First-Approach for Spring for Apache Pulsar based application as that simplifies things tremendously. To encourage this, a `spring-pulsar-spring-boot-starter` module is published that can easily be consumed by an application as a dependency. + +[source,java,indent=0,pending-extract=true,subs="verbatim"] +---- @SpringBootApplication public class PulsarBootHelloWorld { - public static void main(String[] args) { - SpringApplication.run(PulsarBootHelloWorld.class, args); - } + public static void main(String[] args) { + SpringApplication.run(PulsarBootHelloWorld.class, args); + } - @Bean - public ApplicationRunner runner(PulsarTemplate pulsarTemplate) { - return args -> { - for (int i = 0; i < 10; i ++) { - pulsarTemplate.send("hello-pulsar", This is message " + (i + 1)); - } + @Bean + ApplicationRunner runner(PulsarTemplate pulsarTemplate) { + return (args) -> pulsarTemplate.send("hello-pulsar", "Hello Pulsar World!"); + } - }; - } - - @PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") - public void listen(String message) { - System.out.println("Message Received: " + message); - } + @PulsarListener(subscriptionName = "hello-pulsar-subscription", topics = "hello-pulsar") + void listen(String message) { + System.out.println("Message Received: " + message); + } } -``` +---- Let us go through the higher-level details of this application quickly. Later on in this documentation, we will see these components in much more detail. diff --git a/spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java b/spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java index 1b406f0d..893846b1 100644 --- a/spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app1/SpringPulsarBootApp.java @@ -33,7 +33,7 @@ import org.springframework.pulsar.core.PulsarTemplate; @SpringBootApplication public class SpringPulsarBootApp { - Logger logger = LoggerFactory.getLogger(SpringPulsarBootApp.class); + private final Logger logger = LoggerFactory.getLogger(SpringPulsarBootApp.class); public static void main(String[] args) { SpringApplication.run(SpringPulsarBootApp.class, args); @@ -44,7 +44,7 @@ public class SpringPulsarBootApp { * exclusive subscription to consume. */ @Bean - public ApplicationRunner runner1(PulsarTemplate pulsarTemplate) { + ApplicationRunner runner1(PulsarTemplate pulsarTemplate) { String topic1 = "hello-pulsar-exclusive-1"; @@ -56,7 +56,7 @@ public class SpringPulsarBootApp { } @PulsarListener(subscriptionName = "subscription-1", topics = "hello-pulsar-exclusive-1") - public void listen1(String message) { + void listen1(String message) { this.logger.info(message); } @@ -65,7 +65,7 @@ public class SpringPulsarBootApp { * exclusive subscription to consume. */ @Bean - public ApplicationRunner runner2(PulsarTemplate pulsarTemplate) { + ApplicationRunner runner2(PulsarTemplate pulsarTemplate) { String topic1 = "hello-pulsar-exclusive-2"; @@ -77,7 +77,7 @@ public class SpringPulsarBootApp { } @PulsarListener(subscriptionName = "subscription-2", topics = "hello-pulsar-exclusive-2") - public void listen2(Integer message) { + void listen2(Integer message) { this.logger.info("Message received :" + message); } @@ -86,7 +86,7 @@ public class SpringPulsarBootApp { * associated PulsarListener using an exclusive subscription. */ @Bean - public ApplicationRunner runner3(PulsarTemplate pulsarTemplate) { + ApplicationRunner runner3(PulsarTemplate pulsarTemplate) { String topic = "hello-pulsar-exclusive-3"; return args -> { @@ -99,7 +99,7 @@ public class SpringPulsarBootApp { @PulsarListener(subscriptionName = "subscription-3", topics = "hello-pulsar-exclusive-3", schemaType = SchemaType.JSON) - public void listen3(Foo message) { + void listen3(Foo message) { this.logger.info("Message received :" + message); } @@ -107,7 +107,7 @@ public class SpringPulsarBootApp { * Publish and then use PulsarListener in batch listening mode. */ @Bean - public ApplicationRunner runner4(PulsarTemplate pulsarTemplate) { + ApplicationRunner runner4(PulsarTemplate pulsarTemplate) { String topic = "hello-pulsar-exclusive-4"; return args -> { @@ -120,7 +120,7 @@ public class SpringPulsarBootApp { @PulsarListener(subscriptionName = "subscription-4", topics = "hello-pulsar-exclusive-4", schemaType = SchemaType.JSON, batch = true) - public void listen4(List messages) { + void listen4(List messages) { this.logger.info("records received :" + messages.size()); for (Foo message : messages) { this.logger.info("record : " + message); diff --git a/spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java b/spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java index ff7a30d4..df1f71b0 100644 --- a/spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java +++ b/spring-pulsar-sample-apps/src/main/java/app2/FailoverConsumerApp.java @@ -34,14 +34,14 @@ import org.springframework.pulsar.core.PulsarTemplate; @SpringBootApplication public class FailoverConsumerApp { - Logger logger = LoggerFactory.getLogger(FailoverConsumerApp.class); + private final Logger logger = LoggerFactory.getLogger(FailoverConsumerApp.class); public static void main(String[] args) { SpringApplication.run(FailoverConsumerApp.class, "--spring.pulsar.producer.messageRoutingMode=CustomPartition"); } @Bean - public ApplicationRunner runner(PulsarTemplate pulsarTemplate) { + ApplicationRunner runner(PulsarTemplate pulsarTemplate) { String topic = "failover-demo-topic"; return args -> { for (int i = 0; i < 10; i++) { @@ -55,19 +55,19 @@ public class FailoverConsumerApp { @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", subscriptionType = "failover") - public void listen1(String foo) { + void listen1(String foo) { this.logger.info("failover-listen1 : " + foo); } @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", subscriptionType = "failover") - public void listen2(String foo) { + void listen2(String foo) { this.logger.info("failover-listen2 : " + foo); } @PulsarListener(subscriptionName = "failover-subscription-demo", topics = "failover-demo-topic", subscriptionType = "failover") - public void listen(String foo) { + void listen(String foo) { this.logger.info("failover-listen3 : " + foo); }