Add startup failure policy to listeners (#824)

Previously, when a listener container failed to start, it
would only log the exception. This commit introduces
`StartupFailurePolicy` that allows listener containers to CONTINUE,
STOP, RETRY when an error is encountered on startup.

See #445
See #816

Author: onobc

spring-pulsar-docs/src/main/antora/modules/ROOT/pages/reference/message-listener-startup-failure.adoc

@@ -0,0 +1,28 @@
+= Handling Startup Failures
+include::../attributes/attributes-variables.adoc[]
+
+Message listener containers are started when the application context is refreshed.
+By default, any failures encountered during startup are re-thrown and the application will fail to start.
+You can adjust this behavior with the `StartupFailurePolicy` on the corresponding container properties.
+
+The available options are:
+
+- `Stop` (default) - log and re-throw the exception, effectively stopping the application
+- `Continue` - log the exception, leave the container in a non-running state, but do not stop the application
+- `Retry` - log the exception, retry to start the container asynchronously, but do not stop the application.
+
+The default retry behavior is to retry 3 times with a 10-second delay between
+each attempt.
+However, a custom retry template can be specified on the corresponding container properties.
+If the container fails to restart after the retries are exhausted, it is left in a non-running state.
+
+[discrete]
+== Configuration
+
+[discrete]
+=== With Spring Boot
+**TODO**
+
+[discrete]
+=== Without Spring Boot
+**TODO**

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

@@ -951,8 +951,11 @@ The framework detects the provided bean through the `PulsarListener` and applies
 
 If you have multiple `PulsarListener` methods, and each of them have different customization rules, you should create multiple customizer beans and attach the proper customizers on each `PulsarListener`.
 
+[[message-listener-lifecycle]]
+== Message Listener Container Lifecycle
 
-== Pausing and Resuming Message Listener Containers
+[[message-listener-pause-resume]]
+=== Pausing and Resuming
 
 There are situations in which an application might want to pause message consumption temporarily and then resume later.
 Spring for Apache Pulsar provides the ability to pause and resume the underlying message listener containers.
@@ -973,6 +976,10 @@ void someMethod() {
 
 TIP: The id parameter passed to `getListenerContainer` is the container id - which will be the value of the `@PulsarListener` id attribute when pausing/resuming a `@PulsarListener`.
 
+[[message-listener-startup-failure]]
+include::../message-listener-startup-failure.adoc[leveloffset=+2]
+
+
 [[imperative-pulsar-reader]]
 == Pulsar Reader Support
 The framework provides support for using {apache-pulsar-docs}/concepts-clients/#reader-interface[Pulsar Reader] via the `PulsarReaderFactory`.
@@ -1023,3 +1030,6 @@ public PulsarReaderReaderBuilderCustomizer<String> myCustomizer() {
 ----
 
 TIP: If your application only has a single `@PulsarReader` and a single `PulsarReaderReaderBuilderCustomizer` bean registered then the customizer will be automatically applied.
+
+=== Handling Startup Failures
+The same xref:#message-listener-startup-failure[startup failure facilities] available to message listener containers are available for reader containers.

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

@@ -206,6 +206,9 @@ The "listener" aspect is provided by the `ReactivePulsarMessageHandler` of which
 
 NOTE: If topic information is not specified when using the listener containers directly, the same xref:reference/topic-resolution.adoc#topic-resolution-process[topic resolution process] used by the `ReactivePulsarListener` is used with the one exception that the "Message type default" step is **omitted**.
 
+[[message-listener-startup-failure]]
+include::../message-listener-startup-failure.adoc[leveloffset=+2]
+
 [[reactive-concurrency]]
 == Concurrency
 When consuming records in streaming mode (`stream = true`) concurrency comes naturally via the underlying Reactive support in the client implementation.

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/DefaultReactivePulsarMessageListenerContainer.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -19,6 +19,8 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.locks.ReentrantLock;
 
@@ -29,6 +31,8 @@
 import org.apache.pulsar.reactive.client.internal.api.ApiImplementationFactory;
 
 import org.springframework.core.log.LogAccessor;
+import org.springframework.pulsar.PulsarException;
+import org.springframework.pulsar.config.StartupFailurePolicy;
 import org.springframework.pulsar.reactive.core.ReactiveMessageConsumerBuilderCustomizer;
 import org.springframework.pulsar.reactive.core.ReactivePulsarConsumerFactory;
 import org.springframework.util.CollectionUtils;
@@ -38,6 +42,7 @@
  *
  * @param <T> message type.
  * @author Christophe Bornet
+ * @author Chris Bono
  */
 public non-sealed class DefaultReactivePulsarMessageListenerContainer<T>
 		implements ReactivePulsarMessageListenerContainer<T> {
@@ -135,13 +140,50 @@ public void stop() {
 
 	private void doStart() {
 		setRunning(true);
-		this.pipeline = startPipeline(this.pulsarContainerProperties);
+		var containerProps = this.getContainerProperties();
+		try {
+			this.pipeline = startPipeline(this.pulsarContainerProperties);
+		}
+		catch (Exception e) {
+			this.logger.error(e, () -> "Error starting Reactive pipeline");
+			this.doStop();
+			if (containerProps.getStartupFailurePolicy() == StartupFailurePolicy.STOP) {
+				this.logger.info(() -> "Configured to stop on startup failures - exiting");
+				throw new IllegalStateException("Error starting Reactive pipeline", e);
+			}
+		}
+		// Pipeline started w/o errors - short circuit
+		if (this.pipeline != null && this.pipeline.isRunning()) {
+			return;
+		}
+
+		if (containerProps.getStartupFailurePolicy() == StartupFailurePolicy.RETRY) {
+			this.logger.info(() -> "Configured to retry on startup failures - retrying");
+			CompletableFuture.supplyAsync(() -> {
+				var retryTemplate = Optional.ofNullable(containerProps.getStartupFailureRetryTemplate())
+					.orElseGet(containerProps::getDefaultStartupFailureRetryTemplate);
+				return retryTemplate
+					.<ReactiveMessagePipeline, PulsarException>execute((__) -> startPipeline(containerProps));
+			}).whenComplete((p, ex) -> {
+				if (ex == null) {
+					this.pipeline = p;
+					setRunning(this.pipeline != null ? this.pipeline.isRunning() : false);
+				}
+				else {
+					this.logger.error(ex, () -> "Unable to start Reactive pipeline");
+					this.doStop();
+				}
+			});
+		}
 	}
 
 	public void doStop() {
 		try {
 			this.logger.info("Closing Pulsar Reactive pipeline.");
-			this.pipeline.close();
+			if (this.pipeline != null) {
+				this.pipeline.close();
+				this.pipeline = null;
+			}
 		}
 		catch (Exception e) {
 			this.logger.error(e, () -> "Error closing Pulsar Reactive pipeline.");
@@ -174,6 +216,9 @@ private ReactiveMessagePipeline startPipeline(ReactivePulsarContainerProperties<
 			customizers.add(this.consumerCustomizer);
 		}
 
+		// NOTE: The following various pipeline builders always set 'pipelineRetrySpec'
+		// to null as the container controls the retry of the pipeline start. Otherwise
+		// they do not work well together.
 		ReactiveMessageConsumer<T> consumer = getReactivePulsarConsumerFactory()
 			.createConsumer(containerProperties.getSchema(), customizers);
 		ReactiveMessagePipelineBuilder<T> pipelineBuilder = ApiImplementationFactory
@@ -183,6 +228,7 @@ private ReactiveMessagePipeline startPipeline(ReactivePulsarContainerProperties<
 		if (messageHandler instanceof ReactivePulsarStreamingHandler<?>) {
 			pipeline = pipelineBuilder
 				.streamingMessageHandler(((ReactivePulsarStreamingHandler<T>) messageHandler)::received)
+				.pipelineRetrySpec(null)
 				.build();
 		}
 		else {
@@ -195,10 +241,10 @@ private ReactiveMessagePipeline startPipeline(ReactivePulsarContainerProperties<
 				if (containerProperties.isUseKeyOrderedProcessing()) {
 					concurrentPipelineBuilder.useKeyOrderedProcessing();
 				}
-				pipeline = concurrentPipelineBuilder.build();
+				pipeline = concurrentPipelineBuilder.pipelineRetrySpec(null).build();
 			}
 			else {
-				pipeline = pipelineBuilder.build();
+				pipeline = pipelineBuilder.pipelineRetrySpec(null).build();
 			}
 		}
 		pipeline.start();

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/listener/ReactivePulsarContainerProperties.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2022-2023 the original author or authors.
+ * Copyright 2022-2024 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.
@@ -18,16 +18,20 @@
 
 import java.time.Duration;
 import java.util.Collection;
+import java.util.Objects;
 import java.util.regex.Pattern;
 
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionType;
 import org.apache.pulsar.common.schema.SchemaType;
 
+import org.springframework.lang.Nullable;
+import org.springframework.pulsar.config.StartupFailurePolicy;
 import org.springframework.pulsar.core.DefaultSchemaResolver;
 import org.springframework.pulsar.core.DefaultTopicResolver;
 import org.springframework.pulsar.core.SchemaResolver;
 import org.springframework.pulsar.core.TopicResolver;
+import org.springframework.retry.support.RetryTemplate;
 
 /**
  * Contains runtime properties for a reactive listener container.
@@ -61,6 +65,16 @@ public class ReactivePulsarContainerProperties<T> {
 
 	private boolean useKeyOrderedProcessing = false;
 
+	@Nullable
+	private RetryTemplate startupFailureRetryTemplate;
+
+	private final RetryTemplate defaultStartupFailureRetryTemplate = RetryTemplate.builder()
+		.maxAttempts(3)
+		.fixedBackoff(Duration.ofSeconds(10))
+		.build();
+
+	private StartupFailurePolicy startupFailurePolicy = StartupFailurePolicy.STOP;
+
 	public ReactivePulsarMessageHandler getMessageHandler() {
 		return this.messageHandler;
 	}
@@ -161,4 +175,46 @@ public void setUseKeyOrderedProcessing(boolean useKeyOrderedProcessing) {
 		this.useKeyOrderedProcessing = useKeyOrderedProcessing;
 	}
 
+	@Nullable
+	public RetryTemplate getStartupFailureRetryTemplate() {
+		return this.startupFailureRetryTemplate;
+	}
+
+	/**
+	 * Get the default template to use to retry startup when no custom retry template has
+	 * been specified.
+	 * @return the default retry template that will retry 3 times with a fixed delay of 10
+	 * seconds between each attempt.
+	 * @since 1.2.0
+	 */
+	public RetryTemplate getDefaultStartupFailureRetryTemplate() {
+		return this.defaultStartupFailureRetryTemplate;
+	}
+
+	/**
+	 * Set the template to use to retry startup when an exception occurs during startup.
+	 * @param startupFailureRetryTemplate the retry template to use
+	 * @since 1.2.0
+	 */
+	public void setStartupFailureRetryTemplate(RetryTemplate startupFailureRetryTemplate) {
+		this.startupFailureRetryTemplate = startupFailureRetryTemplate;
+		if (this.startupFailureRetryTemplate != null) {
+			setStartupFailurePolicy(StartupFailurePolicy.RETRY);
+		}
+	}
+
+	public StartupFailurePolicy getStartupFailurePolicy() {
+		return this.startupFailurePolicy;
+	}
+
+	/**
+	 * The action to take on the container when a failure occurs during startup.
+	 * @param startupFailurePolicy action to take when a failure occurs during startup
+	 * @since 1.2.0
+	 */
+	public void setStartupFailurePolicy(StartupFailurePolicy startupFailurePolicy) {
+		this.startupFailurePolicy = Objects.requireNonNull(startupFailurePolicy,
+				"startupFailurePolicy must not be null");
+	}
+
 }