Splits Pulsar reference into multiple documents (#627)

This commit splits the large Pulsar reference doc (non-reactive)
into smaller section/pages to improve layout and navigation for
users.

See #463

spring-pulsar-docs/src/main/antora/modules/ROOT/nav.adoc

@@ -6,6 +6,12 @@
 ** xref:intro/getting-help.adoc[]
 * xref:reference/reference.adoc[]
 ** xref:reference/pulsar.adoc[]
+*** xref:reference/pulsar/quick-tour.adoc[]
+*** xref:reference/pulsar/pulsar-client.adoc[]
+*** xref:reference/pulsar/message-production.adoc[]
+*** xref:reference/pulsar/message-consumption.adoc[]
+*** xref:reference/pulsar/topic-resolution.adoc[]
+*** xref:reference/pulsar/publishing-consuming-partitioned-topics.adoc[]
 *** xref:reference/tombstones.adoc[]
 ** xref:reference/reactive-pulsar.adoc[]
 *** xref:reference/tombstones-reactive.adoc[]

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar-admin.adoc

@@ -18,7 +18,7 @@ See the {spring-boot-pulsar-config-props}[`spring.pulsar.admin.*`] application p
 [[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 xref:reference/pulsar.adoc#client-authentication[security configuration] by replacing `spring.pulsar.client` with `spring.pulsar.admin`.
+You can use the aforementioned xref:reference/pulsar/pulsar-client.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/antora/modules/ROOT/pages/reference/pulsar/authentication.adoc

@@ -1,5 +1,4 @@
-
-include::../attributes/attributes.adoc[]
+include::../../attributes/attributes.adoc[]
 
 To connect to a Pulsar cluster that requires authentication, you need to specify which authentication plugin to use and any parameters required by the specified plugin.
 When **using Spring Boot** auto-configuration, you can set the plugin and the plugin parameters via configuration properties (in most cases).
@@ -59,7 +58,7 @@ spring:
           privateKey: ...
           keyId: ...
 ----
-NOTE: This also requires xref:reference/pulsar.adoc#tls-encryption[TLS encryption].
+NOTE: This also requires xref:reference/pulsar/pulsar-client.adoc#tls-encryption[TLS encryption].
 ====
 
 [[Token]]
@@ -135,7 +134,7 @@ spring:
 .[.underline]#Click ##here## for **mTLS (PEM)**#
 [%collapsible]
 ====
-NOTE: Because this option requires TLS encryption, which already requires you to xref:reference/pulsar.adoc#tls-encryption[provide a client builder customizer], it is recommended to simply add the authentication directly on the client builder in your provided TLS customizer.
+NOTE: Because this option requires TLS encryption, which already requires you to xref:reference/pulsar/pulsar-client.adoc#tls-encryption[provide a client builder customizer], it is recommended to simply add the authentication directly on the client builder in your provided TLS customizer.
 You can use the `org.apache.pulsar.client.api.AuthenticationFactory` to help create the authentication object as follows:
 [source,java]
 ----
@@ -148,7 +147,7 @@ See the official Pulsar documentation on {apache-pulsar-docs}/security-tls-authe
 .[.underline]#Click ##here## for **mTLS (JKS)**#
 [%collapsible]
 ====
-NOTE: Because this option requires TLS encryption, which already requires you to xref:reference/pulsar.adoc#tls-encryption[provide a client builder customizer], it is recommended to simply add the authentication directly on the client builder in your provided TLS customizer.
+NOTE: Because this option requires TLS encryption, which already requires you to xref:reference/pulsar/pulsar-client.adoc#tls-encryption[provide a client builder customizer], it is recommended to simply add the authentication directly on the client builder in your provided TLS customizer.
 You can use the `org.apache.pulsar.client.api.AuthenticationFactory` to help create the authentication object as follows:
 [source,java]
 ----

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/message-production.adoc

@@ -0,0 +1,182 @@
+[[message-production]]
+= Message Production
+include::../../attributes/attributes.adoc[]
+
+[[pulsar-producer]]
+== Pulsar Template
+
+On the Pulsar producer side, Spring Boot auto-configuration provides a `PulsarTemplate` for publishing records. The template implements an interface called `PulsarOperations` and provides methods to publish records through its contract.
+
+There are two categories of these send API methods: `send` and `sendAsync`.
+The `send` methods block calls by 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.
+They return a `CompletableFuture`, which you can use to asynchronously receive the message ID once the messages are published.
+
+NOTE: For the API variants that do not include a topic parameter, a xref:reference/pulsar/topic-resolution.adoc#topic-resolution-process-imperative[topic resolution process] is used to determine the destination topic.
+
+=== Simple API
+The template provides a handful of methods ({javadocs}/org/springframework/pulsar/core/PulsarOperations.html[prefixed with _'send'_]) for simple send requests. For more complicated send requests, a fluent API lets you configure more options.
+
+=== Fluent API
+The template provides a {javadocs}/org/springframework/pulsar/core/PulsarOperations.html#newMessage(T)[fluent builder] to handle more complicated send requests.
+
+=== Message customization
+You can specify a `TypedMessageBuilderCustomizer` to configure the outgoing message. For example, the following code shows how to send a keyed message:
+[source, java]
+----
+template.newMessage(msg)
+    .withMessageCustomizer((mb) -> mb.key("foo-msg-key"))
+    .send();
+----
+
+[[single-producer-customize]]
+=== Producer customization
+You can specify a `ProducerBuilderCustomizer` to configure the underlying Pulsar producer builder that ultimately constructs the producer used to send the outgoing message.
+
+WARNING: Use with caution as this gives full access to the producer builder and invoking some of its methods (such as `create`) may have unintended side effects.
+
+For example, the following code shows how to disable batching and enable chunking:
+[source, java]
+----
+template.newMessage(msg)
+    .withProducerCustomizer((pb) -> pb.enableChunking(true).enableBatching(false))
+    .send();
+----
+
+This other example shows how to use custom routing when publishing records to partitioned topics.
+Specify your custom `MessageRouter` implementation on the `Producer` builder such as:
+[source, java]
+----
+template.newMessage(msg)
+    .withProducerCustomizer((pb) -> pb.messageRouter(messageRouter))
+    .send();
+----
+
+TIP: Note that, when using a `MessageRouter`, the only valid setting for `spring.pulsar.producer.message-routing-mode` is `custom`.
+
+This other example shows how to add a `ProducerInterceptor` that will intercept and mutate messages received by the producer before being published to the brokers:
+[source, java]
+----
+template.newMessage(msg)
+    .withProducerCustomizer((pb) -> pb.intercept(interceptor))
+    .send();
+----
+
+The customizer will only apply to the producer used for the send operation.
+If you want to apply a customizer to all producers, you must provide them to the producer factory as described in <<global-producer-customize>>.
+
+CAUTION: The rules described in "`<<producer-caching-lambdas>>`" must be followed when using Lambda customizers.
+
+
+[[schema-info-template-imperative]]
+:template-class: PulsarTemplate
+include::../schema-info/schema-info-template.adoc[]
+
+[[pulsar-producer-factory]]
+== Pulsar Producer Factory
+The `PulsarTemplate` relies on a `PulsarProducerFactory` to actually create the underlying producer.
+Spring Boot auto-configuration also provides this producer factory which you can further configure by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.producer.*`] application properties.
+
+NOTE: If topic information is not specified when using the producer factory APIs directly, the same xref:reference/pulsar/topic-resolution.adoc#topic-resolution-process-imperative[topic resolution process] used by the `PulsarTemplate` is used with the one exception that the "Message type default" step is **omitted**.
+
+[[global-producer-customize]]
+=== Global producer customization
+The framework provides the `ProducerBuilderCustomizer` contract which allows you to configure the underlying builder which is used to construct each producer.
+To customize all producers, you can pass a list of customizers into the `PulsarProducerFactory` constructor.
+When using multiple customizers, they are applied in the order in which they appear in the list.
+
+TIP: If you use Spring Boot auto-configuration, you can specify the customizers as beans and they will be passed automatically to the `PulsarProducerFactory`, ordered according to their `@Order` annotation.
+
+If you want to apply a customizer to just a single producer, you can use the Fluent API and <<single-producer-customize,specify the customizer at send time>>.
+
+[[producer-caching]]
+== Pulsar Producer Caching
+Each underlying Pulsar producer consumes resources.
+To improve performance and avoid continual creation of producers, the producer factory caches the producers that it creates.
+They are cached in an LRU fashion and evicted when they have not been used within a configured time period.
+The link:{github}/blob/8e33ac0b122bc0e75df299919c956cacabcc9809/spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java#L159[cache key] is composed of just enough information to ensure that callers are returned the same producer on subsequent creation requests.
+
+Additionally, you can configure the cache settings by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.producer.cache.*`] application properties.
+
+[[producer-caching-lambdas]]
+=== Caution on Lambda customizers
+Any user-provided producer customizers are also included in the cache key.
+Because the cache key relies on a valid implementation of `equals/hashCode`, one must take caution when using Lambda customizers.
+
+IMPORTANT: *RULE:* Two customizers implemented as Lambdas will match on `equals/hashCode` *if and only if* they use the same Lambda instance and do not require any variable defined outside its closure.
+
+To clarify the above rule we will look at a few examples.
+In the following example, the customizer is defined as an inline Lambda which means that each call to `sendUser` uses the same Lambda instance. Additionally, it requires no variable outside its closure. Therefore, it *will* match as a cache key.
+
+[source, java]
+----
+void sendUser() {
+    var user = randomUser();
+    template.newMessage(user)
+        .withTopic("user-topic")
+        .withProducerCustomizer((b) -> b.producerName("user"))
+        .send();
+}
+----
+
+In this next case, the customizer is defined as an inline Lambda which means that each call to `sendUser` uses the same Lambda instance. However, it requires a variable outside its closure. Therefore, it *will not* match as a cache key.
+
+[source, java]
+----
+void sendUser() {
+    var user = randomUser();
+    var name = randomName();
+    template.newMessage(user)
+        .withTopic("user-topic")
+        .withProducerCustomizer((b) -> b.producerName(name))
+        .send();
+}
+----
+
+In this final example, the customizer is defined as an inline Lambda which means that each call to `sendUser` uses the same Lambda instance. While it does use a variable name, it does not originate outside its closure and therefore *will* match as a cache key.
+This illustrates that variables can be used *within* the Lambda closure and can even make calls to static methods.
+
+[source, java]
+----
+void sendUser() {
+    var user = randomUser();
+    template.newMessage(user)
+        .withTopic("user-topic")
+        .withProducerCustomizer((b) -> {
+           var name = SomeHelper.someStaticMethod();
+           b.producerName(name);
+        })
+        .send();
+}
+----
+
+IMPORTANT: *RULE:* If your Lambda customizer is not defined *once and only once* (the same instance is used on subsequent calls) *OR* it requires variable(s) defined outside its closure then you must provide a customizer implementation with a valid `equals/hashCode` implementation.
+
+WARNING: If these rules are not followed then the producer cache will always miss and your application performance will be negatively affected.
+
+== Intercept Messages on the Producer
+Adding a `ProducerInterceptor`  lets you intercept and mutate messages received by the producer before they are published to the brokers.
+To do so, you can pass a list of interceptors into the `PulsarTemplate` constructor.
+When using multiple interceptors, the order they are applied in is the order in which they appear in the list.
+
+If you use Spring Boot auto-configuration, you can specify the interceptors as Beans.
+They are passed automatically to the `PulsarTemplate`.
+Ordering of the interceptors is achieved by using the `@Order` annotation as follows:
+
+[source, java]
+----
+@Bean
+@Order(100)
+ProducerInterceptor firstInterceptor() {
+  ...
+}
+
+@Bean
+@Order(200)
+ProducerInterceptor secondInterceptor() {
+  ...
+}
+----
+
+NOTE: If you are not using the starter, you will need to configure and register the aforementioned components yourself.

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/preface.adoc

@@ -0,0 +1,9 @@
+[[preface]]
+= Preface
+include::../../attributes/attributes.adoc[]
+
+NOTE: We recommend using a Spring-Boot-First approach for Spring for Apache Pulsar-based applications, as that simplifies things tremendously.
+To do so, you can add the `spring-pulsar-spring-boot-starter` module as a dependency.
+
+NOTE: The majority of this reference expects the reader to be using the starter and gives most directions for configuration with that in mind.
+However, an effort is made to call out when instructions are specific to the Spring Boot starter usage.

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/publishing-consuming-partitioned-topics.adoc

@@ -0,0 +1,111 @@
+[[publishing-consuming-partitioned-topics]]
+= Publishing and Consuming Partitioned Topics
+include::../../attributes/attributes.adoc[]
+
+In the following example, we publish 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 {
+
+	public static void main(String[] args) {
+		SpringApplication.run(PulsarBootPartitioned.class, "--spring.pulsar.producer.message-routing-mode=CustomPartition");
+	}
+
+	@Bean
+	public ApplicationRunner runner(PulsarTemplate<String> pulsarTemplate) {
+		pulsarTemplate.setDefaultTopicName("hello-pulsar-partitioned");
+		return args -> {
+			for (int i = 0; i < 10; i++) {
+				pulsarTemplate.sendAsync("hello john doe 0 ", new FooRouter());
+				pulsarTemplate.sendAsync("hello alice doe 1", new BarRouter());
+				pulsarTemplate.sendAsync("hello buzz doe 2", new BuzzRouter());
+			}
+		};
+	}
+
+	@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned")
+	public void listen(String message) {
+		System.out.println("Message Received: " + message);
+	}
+
+    static class FooRouter implements MessageRouter {
+
+		@Override
+		public int choosePartition(Message<?> msg, TopicMetadata metadata) {
+			return 0;
+		}
+	}
+
+	static class BarRouter implements MessageRouter {
+
+		@Override
+		public int choosePartition(Message<?> msg, TopicMetadata metadata) {
+			return 1;
+		}
+	}
+
+	static class BuzzRouter implements MessageRouter {
+
+		@Override
+		public int choosePartition(Message<?> msg, TopicMetadata metadata) {
+			return 2;
+		}
+	}
+
+}
+----
+
+In the preceding example, we publish to a partitioned topic, and we would like to publish some data segment to a specific partition.
+If you leave it to Pulsar's default, it follows a round-robin mode of partition assignments, and we would like to override that.
+To do so, we provide a message router object with the `send` method.
+Consider the three message routers implemented.
+`FooRouter` always sends data to partition `0`, `BarRouter` sends to partition `1`, and `BuzzRouter` sends to partition `2`.
+Also note that we now use the `sendAsync` method of `PulsarTemplate` that returns a `CompletableFuture`.
+When running the application, we also need to set the `messageRoutingMode` on the producer to `CustomPartition` (`spring.pulsar.producer.message-routing-mode`).
+
+On the consumer side, we use a `PulsarListener` with the exclusive subscription type.
+This means that data from all the partitions ends up in the same consumer and there is no ordering guarantee.
+
+What can we do if we want each partition to be consumed by a single distinct consumer?
+We can switch to the `failover` subscription mode and add three separate consumers:
+
+[source, java]
+----
+@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = SubscriptionType.Failover)
+public void listen1(String foo) {
+    System.out.println("Message Received 1: " + foo);
+}
+
+@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription", topics = "hello-pulsar-partitioned", subscriptionType = SubscriptionType.Failover)
+public void listen2(String foo) {
+    System.out.println("Message Received 2: " + foo);
+}
+
+@PulsarListener(subscriptionName = "hello-pulsar-partitioned-subscription",  topics = "hello-pulsar-partitioned", subscriptionType = SubscriptionType.Failover)
+public void listen3(String foo) {
+    System.out.println("Message Received 3: " + foo);
+}
+----
+
+When you follow this approach, a single partition always gets consumed by a dedicated consumer.
+
+In a similar vein, if you want to use Pulsar's shared consumer type, you can use the `shared` subscription type.
+However, when you use the `shared` mode, you lose any ordering guarantees, as a single consumer may receive messages from all the partitions before another consumer gets a chance.
+
+Consider the following example:
+
+[source, java]
+----
+@PulsarListener(subscriptionName = "hello-pulsar-shared-subscription", topics = "hello-pulsar-partitioned", subscriptionType = SubscriptionType.Shared)
+public void listen1(String foo) {
+    System.out.println("Message Received 1: " + foo);
+}
+
+@PulsarListener(subscriptionName = "hello-pulsar-shared-subscription", topics = "hello-pulsar-partitioned", subscriptionType = SubscriptionType.Shared)
+public void listen2(String foo) {
+    System.out.println("Message Received 2: " + foo);
+}
+----

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/pulsar-client.adoc

@@ -0,0 +1,23 @@
+[[pulsar-client]]
+= Pulsar Client
+include::../../attributes/attributes.adoc[]
+
+When you use the Pulsar Spring Boot Starter, you get the `PulsarClient` auto-configured.
+
+By default, the application tries to connect to a local Pulsar instance at `pulsar://localhost:6650`.
+This can be adjusted by setting the `spring.pulsar.client.service-url` property to a different value.
+
+TIP: The value must be a valid {apache-pulsar-docs}/client-libraries-java/#connection-urls[Pulsar Protocol] URL
+
+You can further configure the client by specifying any of the {spring-boot-pulsar-config-props}[`spring.pulsar.client.*`] application properties.
+
+NOTE: If you are not using the starter, you will need to configure and register the `PulsarClient` yourself.
+There is a `DefaultPulsarClientFactory` that accepts a builder customizer that can be used to help with this.
+
+[[tls-encryption]]
+== TLS Encryption (SSL)
+include::tls-encryption.adoc[]
+
+[[client-authentication]]
+== Authentication
+include::authentication.adoc[]

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/quick-tour.adoc

@@ -1,6 +1,6 @@
 [[quick-tour]]
 = Quick Tour
-include::../attributes/attributes.adoc[]
+include::../../attributes/attributes.adoc[]
 
 We will take a quick tour of Spring for Apache Pulsar by showing a sample Spring Boot application that produces and consumes.
 This is a complete application and does not require any additional configuration, as long as you have a Pulsar cluster running on the default location - `localhost:6650`.

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/tls-encryption.adoc

@@ -1,5 +1,4 @@
-
-include::../attributes/attributes.adoc[]
+include::../../attributes/attributes.adoc[]
 
 By default, Pulsar clients communicate with Pulsar services in plain text.
 The following section describes how to configure Pulsar clients to use TLS encryption (SSL).

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/pulsar/topic-resolution.adoc

@@ -1,3 +1,7 @@
+[[topic-resolution-process-imperative]]
+= Topic Resolution
+include::../../attributes/attributes.adoc[]
+
 A destination topic is needed when producing or consuming messages.
 The framework looks in the following ordered locations to determine a topic (stopping at the first find):
 

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/reactive-pulsar.adoc

@@ -56,7 +56,7 @@ See the {spring-boot-pulsar-config-props}[`spring.pulsar.client.*`] application
 
 [[reactive-client-authentication]]
 === Authentication
-To connect to a Pulsar cluster that requires authentication, follow xref:reference/pulsar.adoc#client-authentication[the same steps] as the imperative client.
+To connect to a Pulsar cluster that requires authentication, follow xref:reference/pulsar/pulsar-client.adoc#client-authentication[the same steps] as the imperative client.
 Again, this is because the reactive client adapts the imperative client which handles all security configuration.
 
 [[reactive-message-production]]
@@ -498,4 +498,4 @@ Spring Boot provides this reader factory which can be configured with any of the
 
 [[topic-resolution-process-reactive]]
 == Topic Resolution
-include::topic-resolution.adoc[leveloffset=+1]
+include::pulsar/topic-resolution.adoc[leveloffset=+1]

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc

@@ -8,13 +8,13 @@ This section covers the changes made from version 1.0 to version 1.1.
 
 === Auto Schema support
 If there is no chance to know the schema of a Pulsar topic in advance, you can use AUTO Schemas to produce/consume generic records to/from brokers.
-See xref:./reference/pulsar.adoc#template-auto-produce[Producing with AUTO_SCHEMA] and xref:./reference/pulsar.adoc#listener-auto-consume[Consuming with AUTO_SCHEMA] for more details.
+See xref:./reference/pulsar/message-production.adoc#template-auto-produce[Producing with AUTO_SCHEMA] and xref:./reference/pulsar/message-consumption.adoc#listener-auto-consume[Consuming with AUTO_SCHEMA] for more details.
 
 NOTE: While the above links focus on `PulsarTemplate` and `@PulsarListener`, this feature is also supported in `ReactivePulsarTemplate`, `@ReactivePulsarListener`, and `@PulsarReader`.
 Details for each can be found in their respective section of this reference guide.
 
 === Default topic/schema via message annotation
-You can now mark a message class with `@PulsarMessage` to specify the xref:./reference/pulsar.adoc#default-topic-via-annotation[default topic] and/or xref:./reference/pulsar.adoc#listener-default-schema-annotation[default schema] to use when producing/consuming messages of that type.
+You can now mark a message class with `@PulsarMessage` to specify the xref:./reference/pulsar/topic-resolution.adoc#default-topic-via-annotation[default topic] and/or xref:./reference/pulsar/message-consumption.adoc#listener-default-schema-annotation[default schema] to use when producing/consuming messages of that type.
 
 === Remove checked exceptions
 The APIs provided by the framework no longer throw the checked `PulsarClientException`, but rather the unchecked `PulsarException`.