Add graceful restart to client (#452)

This commit introduces a restartable client that
participates in SmartLifecycle and handles stop
and start gracefully.

Resolves #422

build.gradle

@@ -34,6 +34,7 @@ nohttp {
 	source.exclude "**/build/**"
 	source.exclude "**/out/**"
 	source.exclude "**/target/**"
+	source.exclude "**/*.dylib"
 }
 
 check {

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactory.java

@@ -19,6 +19,7 @@ package org.springframework.pulsar.reactive.core;
 import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.atomic.AtomicReference;
 
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Schema;
@@ -31,6 +32,7 @@ import org.apache.pulsar.reactive.client.api.ReactivePulsarClient;
 import org.springframework.core.log.LogAccessor;
 import org.springframework.lang.Nullable;
 import org.springframework.pulsar.core.DefaultTopicResolver;
+import org.springframework.pulsar.core.PulsarClientProxy;
 import org.springframework.pulsar.core.TopicResolver;
 import org.springframework.util.Assert;
 import org.springframework.util.CollectionUtils;
@@ -42,10 +44,15 @@ import org.springframework.util.CollectionUtils;
  * @author Christophe Bornet
  * @author Chris Bono
  */
-public final class DefaultReactivePulsarSenderFactory<T> implements ReactivePulsarSenderFactory<T> {
+public final class DefaultReactivePulsarSenderFactory<T>
+		implements ReactivePulsarSenderFactory<T>, RestartableComponentSupport {
+
+	private static final int LIFECYCLE_PHASE = (Integer.MIN_VALUE / 2) - 100;
 
 	private final LogAccessor logger = new LogAccessor(this.getClass());
 
+	private final AtomicReference<State> currentState = RestartableComponentSupport.initialState();
+
 	private final ReactivePulsarClient reactivePulsarClient;
 
 	private final TopicResolver topicResolver;
@@ -110,6 +117,9 @@ public final class DefaultReactivePulsarSenderFactory<T> implements ReactivePuls
 	private ReactiveMessageSender<T> doCreateReactiveMessageSender(Schema<T> schema, @Nullable String topic,
 			@Nullable List<ReactiveMessageSenderBuilderCustomizer<T>> customizers) {
 		Objects.requireNonNull(schema, "Schema must be specified");
+
+		this.logger.warn(() -> "**** Du CreateMessageSender for topic=" + topic);
+
 		String resolvedTopic = this.topicResolver.resolveTopic(topic, () -> getDefaultTopic()).orElseThrow();
 		this.logger.trace(() -> "Creating reactive message sender for '%s' topic".formatted(resolvedTopic));
 
@@ -139,6 +149,41 @@ public final class DefaultReactivePulsarSenderFactory<T> implements ReactivePuls
 		return this.defaultTopic;
 	}
 
+	/**
+	 * Return the phase that this lifecycle object is supposed to run in.
+	 * <p>
+	 * This component has a phase that comes after the {@link PulsarClientProxy
+	 * restartable client} but before other lifecycle and smart lifecycle components whose
+	 * phase values are &quot;0&quot; and &quot;max&quot;, respectively.
+	 * @return a phase that is after the restartable client and before other default
+	 * components.
+	 * @see PulsarClientProxy#getPhase()
+	 */
+	@Override
+	public int getPhase() {
+		return LIFECYCLE_PHASE;
+	}
+
+	@Override
+	public AtomicReference<State> currentState() {
+		return this.currentState;
+	}
+
+	@Override
+	public LogAccessor logger() {
+		return this.logger;
+	}
+
+	@Override
+	public void doStop() {
+		try {
+			this.reactiveMessageSenderCache.close();
+		}
+		catch (Exception e) {
+			throw new RuntimeException(e);
+		}
+	}
+
 	/**
 	 * Builder for {@link DefaultReactivePulsarSenderFactory}.
 	 *

spring-pulsar-reactive/src/main/java/org/springframework/pulsar/reactive/core/RestartableComponentSupport.java

@@ -0,0 +1,138 @@
+/*
+ * Copyright 2022-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.pulsar.reactive.core;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.lang.Nullable;
+
+/**
+ * Provides a simple base implementation for a component that can be restarted (stopped
+ * then started) and still be in a usable state.
+ * <p>
+ * This is an interface that provides default methods that rely on the current component
+ * state which must be maintained by the implementing component.
+ * <p>
+ * This can serve as a base implementation for coordinated checkpoint and restore by
+ * simply implementing the {@link #doStart() start} and/or {@link #doStop() stop} callback
+ * to re-acquire and release resources, respectively.
+ * <p>
+ * Implementors are required to provide the component state and a logger.
+ *
+ * @author Chris Bono
+ */
+interface RestartableComponentSupport extends SmartLifecycle, DisposableBean {
+
+	/**
+	 * Gets the initial state for the implementing component.
+	 * @return the initial component state
+	 */
+	static AtomicReference<State> initialState() {
+		return new AtomicReference<>(State.CREATED);
+	}
+
+	/**
+	 * Callback to get the current state from the component.
+	 * @return the current state of the component
+	 */
+	AtomicReference<State> currentState();
+
+	/**
+	 * Callback to get the component specific logger.
+	 * @return the component specific logger
+	 */
+	LogAccessor logger();
+
+	/**
+	 * Lifecycle state of this factory.
+	 */
+	enum State {
+
+		/** Component initially created. */
+		CREATED,
+		/** Component in the process of being started. */
+		STARTING,
+		/** Component has been started. */
+		STARTED,
+		/** Component in the process of being stopped. */
+		STOPPING,
+		/** Component has been stopped. */
+		STOPPED,
+		/** Component has been destroyed. */
+		DESTROYED;
+
+	}
+
+	@Override
+	default boolean isRunning() {
+		return State.STARTED.equals(currentState().get());
+	}
+
+	@Override
+	default void start() {
+		State current = currentState().getAndUpdate(state -> isCreatedOrStopped(state) ? State.STARTING : state);
+		if (isCreatedOrStopped(current)) {
+			logger().debug(() -> "Starting...");
+			doStart();
+			currentState().set(State.STARTED);
+			logger().debug(() -> "Started");
+		}
+	}
+
+	private static boolean isCreatedOrStopped(@Nullable State state) {
+		return State.CREATED.equals(state) || State.STOPPED.equals(state);
+	}
+
+	/**
+	 * Callback invoked during startup - default implementation does nothing.
+	 */
+	default void doStart() {
+	}
+
+	@Override
+	default void stop() {
+		State current = currentState().getAndUpdate(state -> isCreatedOrStarted(state) ? State.STOPPING : state);
+		if (isCreatedOrStarted(current)) {
+			logger().debug(() -> "Stopping...");
+			doStop();
+			currentState().set(State.STOPPED);
+			logger().debug(() -> "Stopped");
+		}
+	}
+
+	private static boolean isCreatedOrStarted(@Nullable State state) {
+		return State.CREATED.equals(state) || State.STARTED.equals(state);
+	}
+
+	/**
+	 * Callback invoked during stop - default implementation does nothing.
+	 */
+	default void doStop() {
+	}
+
+	@Override
+	default void destroy() {
+		logger().debug(() -> "Destroying...");
+		stop();
+		currentState().set(State.DESTROYED);
+		logger().debug(() -> "Destroyed");
+	}
+
+}

spring-pulsar-reactive/src/test/java/org/springframework/pulsar/reactive/core/DefaultReactivePulsarSenderFactoryTests.java

@@ -20,8 +20,12 @@ import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
 import static org.assertj.core.api.Assertions.assertThatNullPointerException;
 import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.clearInvocations;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
 
 import java.util.Arrays;
 import java.util.Collections;
@@ -210,4 +214,25 @@ class DefaultReactivePulsarSenderFactoryTests {
 
 	}
 
+	@Nested
+	class RestartFactoryTests {
+
+		@Test
+		void restartLifecycle() throws Exception {
+			var cache = spy(AdaptedReactivePulsarClientFactory.createCache());
+			var senderFactory = (DefaultReactivePulsarSenderFactory<String>) newSenderFactoryWithCache(cache);
+			senderFactory.start();
+			senderFactory.createSender(schema, "topic1");
+			senderFactory.stop();
+			senderFactory.stop();
+			verify(cache, times(1)).close();
+			clearInvocations(cache);
+			senderFactory.start();
+			senderFactory.createSender(schema, "topic2");
+			senderFactory.stop();
+			verify(cache, times(1)).close();
+		}
+
+	}
+
 }

spring-pulsar/src/main/java/org/springframework/pulsar/core/CachingPulsarProducerFactory.java

@@ -23,6 +23,7 @@ import java.util.List;
 import java.util.Objects;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 import org.apache.pulsar.client.api.MessageId;
@@ -58,12 +59,17 @@ import org.springframework.util.Assert;
  * @author Alexander Preuß
  * @author Christophe Bornet
  */
-public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactory<T> implements DisposableBean {
+public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactory<T>
+		implements RestartableComponentSupport {
+
+	private static final int LIFECYCLE_PHASE = (Integer.MIN_VALUE / 2) - 100;
 
 	private final LogAccessor logger = new LogAccessor(this.getClass());
 
 	private final CacheProvider<ProducerCacheKey<T>, Producer<T>> producerCache;
 
+	private final AtomicReference<State> currentState = RestartableComponentSupport.initialState();
+
 	/**
 	 * Construct a caching producer factory with the specified values for the cache
 	 * configuration.
@@ -85,7 +91,7 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 				(key, producer, cause) -> {
 					this.logger.debug(() -> "Producer %s evicted from cache due to %s"
 						.formatted(ProducerUtils.formatProducer(producer), cause));
-					closeProducer(producer);
+					closeProducer(producer, true);
 				});
 	}
 
@@ -113,22 +119,51 @@ public class CachingPulsarProducerFactory<T> extends DefaultPulsarProducerFactor
 		}
 	}
 
+	/**
+	 * Return the phase that this lifecycle object is supposed to run in.
+	 * <p>
+	 * Because this object depends on the restartable client, it uses a phase slightly
+	 * larger than the one used by the restartable client. This ensures that it starts
+	 * after and stops before the restartable client.
+	 * @return the phase to execute in (just after the restartable client)
+	 * @see PulsarClientProxy#getPhase()
+	 */
 	@Override
-	public void destroy() {
-		this.producerCache.invalidateAll((key, producer) -> closeProducer(producer));
+	public int getPhase() {
+		return LIFECYCLE_PHASE;
 	}
 
-	private void closeProducer(Producer<T> producer) {
+	@Override
+	public AtomicReference<State> currentState() {
+		return this.currentState;
+	}
+
+	@Override
+	public LogAccessor logger() {
+		return this.logger;
+	}
+
+	@Override
+	public void doStop() {
+		this.producerCache.invalidateAll((key, producer) -> closeProducer(producer, false));
+	}
+
+	private void closeProducer(Producer<T> producer, boolean async) {
 		Producer<T> actualProducer = null;
 		if (producer instanceof ProducerWithCloseCallback<T> wrappedProducer) {
 			actualProducer = wrappedProducer.getActualProducer();
 		}
 		if (actualProducer == null) {
-			this.logger.warn(() -> "Unable to get actual producer for %s - will skip closing it"
+			this.logger.trace(() -> "Unable to get actual producer for %s - will skip closing it"
 				.formatted(ProducerUtils.formatProducer(producer)));
 			return;
 		}
-		ProducerUtils.closeProducerAsync(actualProducer, this.logger);
+		if (async) {
+			ProducerUtils.closeProducerAsync(actualProducer, this.logger);
+		}
+		else {
+			ProducerUtils.closeProducer(actualProducer, this.logger, Duration.ofSeconds(15L));
+		}
 	}
 
 	/**

spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarClientFactory.java

@@ -19,6 +19,9 @@ package org.springframework.pulsar.core;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
 
+import org.springframework.context.EnvironmentAware;
+import org.springframework.core.env.Environment;
+import org.springframework.core.log.LogAccessor;
 import org.springframework.util.Assert;
 
 /**
@@ -27,10 +30,14 @@ import org.springframework.util.Assert;
  * @author Soby Chacko
  * @author Chris Bono
  */
-public class DefaultPulsarClientFactory implements PulsarClientFactory {
+public class DefaultPulsarClientFactory implements PulsarClientFactory, EnvironmentAware {
+
+	private final LogAccessor logger = new LogAccessor(this.getClass());
 
 	private final PulsarClientBuilderCustomizer customizer;
 
+	private boolean useRestartableClient;
+
 	/**
 	 * Construct a factory that creates clients using a default Pulsar client builder with
 	 * no modifications other than the specified service url.
@@ -51,9 +58,18 @@ public class DefaultPulsarClientFactory implements PulsarClientFactory {
 
 	@Override
 	public PulsarClient createClient() throws PulsarClientException {
+		if (this.useRestartableClient) {
+			this.logger.info(() -> "Using restartable client");
+			return new PulsarClientProxy(this.customizer);
+		}
 		var clientBuilder = PulsarClient.builder();
 		this.customizer.customize(clientBuilder);
 		return clientBuilder.build();
 	}
 
+	@Override
+	public void setEnvironment(Environment environment) {
+		this.useRestartableClient = environment.getProperty("spring.pulsar.client.restartable", Boolean.class, true);
+	}
+
 }

spring-pulsar/src/main/java/org/springframework/pulsar/core/ProducerUtils.java

@@ -16,6 +16,10 @@
 
 package org.springframework.pulsar.core;
 
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
 import org.apache.pulsar.client.api.Producer;
 
 import org.springframework.core.log.LogAccessor;
@@ -34,11 +38,20 @@ final class ProducerUtils {
 		return "(%s:%s)".formatted(producer.getProducerName(), producer.getTopic());
 	}
 
-	static <T> void closeProducerAsync(Producer<T> producer, LogAccessor logger) {
-		producer.closeAsync().exceptionally(e -> {
+	static <T> CompletableFuture<Void> closeProducerAsync(Producer<T> producer, LogAccessor logger) {
+		return producer.closeAsync().exceptionally(e -> {
 			logger.warn(e, () -> "Failed to close producer %s".formatted(ProducerUtils.formatProducer(producer)));
 			return null;
 		});
 	}
 
+	static <T> void closeProducer(Producer<T> producer, LogAccessor logger, Duration maxWaitTime) {
+		try {
+			producer.closeAsync().get(maxWaitTime.toMillis(), TimeUnit.MILLISECONDS);
+		}
+		catch (Exception e) {
+			logger.warn(e, () -> "Failed to close producer %s".formatted(ProducerUtils.formatProducer(producer)));
+		}
+	}
+
 }

spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarClientProxy.java

@@ -0,0 +1,203 @@
+/*
+ * Copyright 2022-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.pulsar.core;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+
+import org.apache.pulsar.client.api.ConsumerBuilder;
+import org.apache.pulsar.client.api.ProducerBuilder;
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.ReaderBuilder;
+import org.apache.pulsar.client.api.Schema;
+import org.apache.pulsar.client.api.TableViewBuilder;
+import org.apache.pulsar.client.api.transaction.TransactionBuilder;
+
+import org.springframework.context.Lifecycle;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.util.Assert;
+
+/**
+ * A {@link PulsarClient} implementation that delegates to another actual Pulsar client.
+ * The proxy client can be stopped and then started and still be in a usable state without
+ * any knowledge of the restart or changes required to the users of the client.
+ * <p>
+ * The proxy client participates in the Spring {@link SmartLifecycle Lifecycle} and closes
+ * the underlying client when {@link SmartLifecycle#stop() stopped} and creates another
+ * delegate client when subsequently {@link SmartLifecycle#start() started}.
+ *
+ * @author Chris Bono
+ */
+public final class PulsarClientProxy extends RestartableSingletonFactory<PulsarClient> implements PulsarClient {
+
+	private static final int LIFECYCLE_PHASE = (Integer.MIN_VALUE / 2) - 200;
+
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
+	private final PulsarClientBuilderCustomizer customizer;
+
+	/**
+	 * Construct a factory that creates clients using a customized Pulsar client builder.
+	 * @param customizer the customizer to apply to the builder
+	 */
+	PulsarClientProxy(PulsarClientBuilderCustomizer customizer) {
+		Assert.notNull(customizer, "customizer must not be null");
+		this.customizer = customizer;
+	}
+
+	/**
+	 * Return the phase that this lifecycle object is supposed to run in.
+	 * <p>
+	 * Lifecycle objects are started in ascending phase order (those w/ smaller phases are
+	 * started before those with larger phases).
+	 * <p>
+	 * Lifecycle objects are stopped in descending phase order (those w/ larger phases are
+	 * stopped before those with smaller phases).
+	 * <p>
+	 * The phases range from {@link Integer#MIN_VALUE} to {@link Integer#MAX_VALUE}.
+	 * <p>
+	 * The restartable client has a phase value that is roughly at the half-way marker on
+	 * the left hand side of the phase continuum (in the middle of &quot;min&quot; and
+	 * &quot;0&quot;). If another component depends on the restartable client it should
+	 * use a phase that is greater than this value.
+	 * <p>
+	 * Because {@link Lifecycle regular} lifecycle objects have a default phase of
+	 * &quot;0&quot; and {@link SmartLifecycle smart} lifecycle objects have a default
+	 * phase of &quot;max&quot;, the restartable client will be started before (and
+	 * stopped after) the majority of all other default configured lifecycle objects.
+	 * @return the phase to execute in ({@link #LIFECYCLE_PHASE})
+	 */
+	@Override
+	public int getPhase() {
+		return LIFECYCLE_PHASE;
+	}
+
+	@Override
+	protected PulsarClient createInstance() {
+		this.logger.debug(() -> "Creating client");
+		var clientBuilder = PulsarClient.builder();
+		this.customizer.customize(clientBuilder);
+		try {
+			return clientBuilder.build();
+		}
+		catch (PulsarClientException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	@Override
+	protected void stopInstance(PulsarClient pulsarClient) {
+		this.logger.debug(() -> "Closing client");
+		try {
+			pulsarClient.close();
+		}
+		catch (PulsarClientException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	@Override
+	protected boolean discardInstanceAfterStop() {
+		return false;
+	}
+
+	// --- PulsarClient implementation below here
+
+	@Override
+	public ProducerBuilder<byte[]> newProducer() {
+		return this.getInstance().newProducer();
+	}
+
+	@Override
+	public <T> ProducerBuilder<T> newProducer(Schema<T> schema) {
+		return this.getInstance().newProducer(schema);
+	}
+
+	@Override
+	public ConsumerBuilder<byte[]> newConsumer() {
+		return this.getInstance().newConsumer();
+	}
+
+	@Override
+	public <T> ConsumerBuilder<T> newConsumer(Schema<T> schema) {
+		return this.getInstance().newConsumer(schema);
+	}
+
+	@Override
+	public ReaderBuilder<byte[]> newReader() {
+		return this.getInstance().newReader();
+	}
+
+	@Override
+	public <T> ReaderBuilder<T> newReader(Schema<T> schema) {
+		return this.getInstance().newReader(schema);
+	}
+
+	@SuppressWarnings("deprecation")
+	@Override
+	public <T> TableViewBuilder<T> newTableViewBuilder(Schema<T> schema) {
+		return this.getInstance().newTableViewBuilder(schema);
+	}
+
+	@Override
+	public TableViewBuilder<byte[]> newTableView() {
+		return this.getInstance().newTableView();
+	}
+
+	@Override
+	public <T> TableViewBuilder<T> newTableView(Schema<T> schema) {
+		return this.getInstance().newTableView(schema);
+	}
+
+	@Override
+	public void updateServiceUrl(String serviceUrl) throws PulsarClientException {
+		this.getInstance().updateServiceUrl(serviceUrl);
+	}
+
+	@Override
+	public CompletableFuture<List<String>> getPartitionsForTopic(String topic) {
+		return this.getInstance().getPartitionsForTopic(topic);
+	}
+
+	@Override
+	public void close() throws PulsarClientException {
+		this.getInstance().close();
+	}
+
+	@Override
+	public CompletableFuture<Void> closeAsync() {
+		return this.getInstance().closeAsync();
+	}
+
+	@Override
+	public void shutdown() throws PulsarClientException {
+		this.getInstance().shutdown();
+	}
+
+	@Override
+	public boolean isClosed() {
+		return this.getInstance().isClosed();
+	}
+
+	@Override
+	public TransactionBuilder newTransaction() {
+		return this.getInstance().newTransaction();
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentBase.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2022-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.pulsar.core;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.springframework.core.log.LogAccessor;
+
+/**
+ * Provides a simple base implementation for a component that can be restarted (stopped
+ * then started) and still be in a usable state.
+ * <p>
+ * Subclasses can use this as a base implementation for coordinated checkpoint and restore
+ * by simply implementing the {@link #doStart() start} and/or {@link #doStop() stop}
+ * callback to re-acquire and release resources, respectively.
+ *
+ * @author Chris Bono
+ */
+abstract class RestartableComponentBase implements RestartableComponentSupport {
+
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
+	private final AtomicReference<State> state = RestartableComponentSupport.initialState();
+
+	@Override
+	public AtomicReference<State> currentState() {
+		return this.state;
+	}
+
+	@Override
+	public LogAccessor logger() {
+		return this.logger;
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableComponentSupport.java

@@ -0,0 +1,143 @@
+/*
+ * Copyright 2022-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.pulsar.core;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.lang.Nullable;
+
+/**
+ * Provides a simple base implementation for a component that can be restarted (stopped
+ * then started) and still be in a usable state.
+ * <p>
+ * This is an interface that provides default methods that rely on the current component
+ * state which must be maintained by the implementing component.
+ * <p>
+ * This can serve as a base implementation for coordinated checkpoint and restore by
+ * simply implementing the {@link #doStart() start} and/or {@link #doStop() stop} callback
+ * to re-acquire and release resources, respectively.
+ * <p>
+ * Implementors are required to provide the component state and an optional logger - see
+ * {@link RestartableComponentBase} for an example.
+ *
+ * @author Chris Bono
+ * @see RestartableComponentBase
+ */
+interface RestartableComponentSupport extends SmartLifecycle, DisposableBean {
+
+	/** Trace logger used for method invocations. */
+	LogAccessor logger = new LogAccessor(RestartableComponentSupport.class);
+
+	/**
+	 * Gets the initial state for the implementing component.
+	 * @return the initial component state
+	 */
+	static AtomicReference<State> initialState() {
+		return new AtomicReference<>(State.CREATED);
+	}
+
+	/**
+	 * Callback to get the current state from the component.
+	 * @return the current state of the component
+	 */
+	AtomicReference<State> currentState();
+
+	/**
+	 * Callback to get the component specific logger.
+	 * @return the component specific logger
+	 */
+	LogAccessor logger();
+
+	/**
+	 * Lifecycle state of this factory.
+	 */
+	enum State {
+
+		/** Component initially created. */
+		CREATED,
+		/** Component in the process of being started. */
+		STARTING,
+		/** Component has been started. */
+		STARTED,
+		/** Component in the process of being stopped. */
+		STOPPING,
+		/** Component has been stopped. */
+		STOPPED,
+		/** Component has been destroyed. */
+		DESTROYED;
+
+	}
+
+	@Override
+	default boolean isRunning() {
+		return State.STARTED.equals(currentState().get());
+	}
+
+	@Override
+	default void start() {
+		State current = currentState().getAndUpdate(state -> isCreatedOrStopped(state) ? State.STARTING : state);
+		if (isCreatedOrStopped(current)) {
+			logger().debug(() -> "Starting...");
+			doStart();
+			currentState().set(State.STARTED);
+			logger().debug(() -> "Started");
+		}
+	}
+
+	private static boolean isCreatedOrStopped(@Nullable State state) {
+		return State.CREATED.equals(state) || State.STOPPED.equals(state);
+	}
+
+	/**
+	 * Callback invoked during startup - default implementation does nothing.
+	 */
+	default void doStart() {
+	}
+
+	@Override
+	default void stop() {
+		State current = currentState().getAndUpdate(state -> isCreatedOrStarted(state) ? State.STOPPING : state);
+		if (isCreatedOrStarted(current)) {
+			logger().debug(() -> "Stopping...");
+			doStop();
+			currentState().set(State.STOPPED);
+			logger().debug(() -> "Stopped");
+		}
+	}
+
+	private static boolean isCreatedOrStarted(@Nullable State state) {
+		return State.CREATED.equals(state) || State.STARTED.equals(state);
+	}
+
+	/**
+	 * Callback invoked during stop - default implementation does nothing.
+	 */
+	default void doStop() {
+	}
+
+	@Override
+	default void destroy() {
+		logger().debug(() -> "Destroying...");
+		stop();
+		currentState().set(State.DESTROYED);
+		logger().debug(() -> "Destroyed");
+	}
+
+}

spring-pulsar/src/main/java/org/springframework/pulsar/core/RestartableSingletonFactory.java

@@ -0,0 +1,147 @@
+/*
+ * Copyright 2022-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.pulsar.core;
+
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.context.SmartLifecycle;
+import org.springframework.core.log.LogAccessor;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * Provides a simple base implementation for a restartable singleton factory.
+ * <p>
+ * It is restartable in the sense that it can be stopped and then started and still be in
+ * a usable state.
+ * <p>
+ * Because it releases its resources when {@link SmartLifecycle#stop() stopped} and
+ * re-acquires them when subsequently {@link SmartLifecycle#start() started}, it can also
+ * be used as a base implementation for coordinated checkpoint and restore.
+ *
+ * @param <T> the bean type
+ * @author Chris Bono
+ */
+abstract class RestartableSingletonFactory<T> extends RestartableComponentBase implements InitializingBean {
+
+	private final LogAccessor logger = new LogAccessor(this.getClass());
+
+	private final AtomicBoolean initialized = new AtomicBoolean(false);
+
+	private T instance;
+
+	protected RestartableSingletonFactory() {
+		super();
+	}
+
+	protected RestartableSingletonFactory(T instance) {
+		super();
+		Assert.notNull(instance, () -> "instance must not be null");
+		this.instance = instance;
+		this.initialized.set(true);
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+		ensureInstanceCreated();
+	}
+
+	@Override
+	public void doStart() {
+		ensureInstanceCreated();
+	}
+
+	private void ensureInstanceCreated() {
+		if (this.initialized.compareAndSet(false, true)) {
+			this.logger.debug(() -> "Creating instance");
+			this.instance = createInstance();
+		}
+	}
+
+	@Override
+	public void doStop() {
+		if (this.instance != null) {
+			this.logger.debug(() -> "Stopping instance");
+			stopInstance(this.instance);
+			if (this.discardInstanceAfterStop()) {
+				this.logger.debug(() -> "Discarding instance");
+				this.instance = null;
+			}
+		}
+		this.initialized.set(false);
+	}
+
+	@Override
+	public void destroy() {
+		super.destroy();
+		this.instance = null;
+	}
+
+	/**
+	 * Gets the singleton instance.
+	 * @return the singleton instance
+	 */
+	public final T getInstance() {
+		return this.instance;
+	}
+
+	/**
+	 * Template method that subclasses must override to construct the backing singleton
+	 * instance returned by this factory.
+	 * <p>
+	 * Implementations should throw a {@link RuntimeException} if an error occurs during
+	 * creation.
+	 * <p>
+	 * Invoked on {@link #afterPropertiesSet() initialization} of this bean if the
+	 * instance has not already been set via the constructor OR during {@link #start()} if
+	 * the instance is null.
+	 * @return the single object managed by the factory
+	 */
+	protected abstract T createInstance();
+
+	/**
+	 * Callback to allow the singleton instance to be &quot;stopped&quot; (ie. allow it to
+	 * release any resources).
+	 * <p>
+	 * Implementations should throw a {@link RuntimeException} if an error occurs during
+	 * destruction.
+	 * <p>
+	 * The default implementation is empty.
+	 * @param instance the singleton instance, as returned by {@link #createInstance()}
+	 */
+	protected void stopInstance(@Nullable T instance) {
+	}
+
+	/**
+	 * Whether to discard the singleton instance (set reference to it null) when stopped
+	 * (default is true).
+	 * @return whether to discard the singleton when stopped
+	 */
+	protected boolean discardInstanceAfterStop() {
+		return true;
+	}
+
+	/**
+	 * Whether the singleton instance has been initialized.
+	 * @return whether the singleton instance has been initialized
+	 */
+	protected boolean initialized() {
+		return this.initialized.get();
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/core/CachingPulsarProducerFactoryTests.java

@@ -230,8 +230,8 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 	@Override
 	protected CachingPulsarProducerFactory<String> producerFactory(PulsarClient pulsarClient,
 			@Nullable String defaultTopic, @Nullable List<ProducerBuilderCustomizer<String>> defaultConfigCustomizers) {
-		var producerFactory = new CachingPulsarProducerFactory<String>(pulsarClient, defaultTopic,
-				defaultConfigCustomizers, new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2);
+		var producerFactory = new CachingPulsarProducerFactory<>(pulsarClient, defaultTopic, defaultConfigCustomizers,
+				new DefaultTopicResolver(), Duration.ofMinutes(5L), 30L, 2);
 		producerFactories.add(producerFactory);
 		return producerFactory;
 	}
@@ -311,4 +311,25 @@ class CachingPulsarProducerFactoryTests extends PulsarProducerFactoryTests {
 
 	}
 
+	@Nested
+	class RestartFactoryTests {
+
+		@Test
+		void restartLifecycle() throws PulsarClientException {
+			var producerFactory = (CachingPulsarProducerFactory<String>) producerFactory(pulsarClient, null, null);
+			producerFactory.start();
+			var producer1 = producerFactory.createProducer(schema, "topic1");
+			var producer2 = producerFactory.createProducer(schema, "topic2");
+			assertThat(actualProducer(producer1).isConnected()).isTrue();
+			assertThat(actualProducer(producer2).isConnected()).isTrue();
+			producerFactory.stop();
+			assertThat(actualProducer(producer1).isConnected()).isFalse();
+			assertThat(actualProducer(producer2).isConnected()).isFalse();
+			producerFactory.start();
+			var producer3 = producerFactory.createProducer(schema, "topic3");
+			assertThat(actualProducer(producer3).isConnected()).isTrue();
+		}
+
+	}
+
 }

spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarClientFactoryTests.java

@@ -23,6 +23,8 @@ import static org.assertj.core.api.Assertions.assertThatRuntimeException;
 import org.apache.pulsar.client.api.PulsarClientException;
 import org.junit.jupiter.api.Test;
 
+import org.springframework.mock.env.MockEnvironment;
+
 /**
  * Tests for {@link DefaultPulsarClientFactory}.
  *
@@ -60,4 +62,27 @@ class DefaultPulsarClientFactoryTests {
 		assertThatRuntimeException().isThrownBy(clientFactory::createClient).withMessage("Who turned out the lights?");
 	}
 
+	@Test
+	void createsRestartableClientByDefault() throws PulsarClientException {
+		var clientFactory = new DefaultPulsarClientFactory("pulsar://localhost:5150");
+		clientFactory.setEnvironment(new MockEnvironment());
+		assertThat(clientFactory.createClient()).isInstanceOf(PulsarClientProxy.class);
+	}
+
+	@Test
+	void createsRestartableClientWhenPropertySetTrue() throws PulsarClientException {
+		var clientFactory = new DefaultPulsarClientFactory("pulsar://localhost:5150");
+		var env = new MockEnvironment().withProperty("spring.pulsar.client.restartable", "true");
+		clientFactory.setEnvironment(env);
+		assertThat(clientFactory.createClient()).isInstanceOf(PulsarClientProxy.class);
+	}
+
+	@Test
+	void createsDefaultClientWhenPropertySetFalse() throws PulsarClientException {
+		var clientFactory = new DefaultPulsarClientFactory("pulsar://localhost:5150");
+		var env = new MockEnvironment().withProperty("spring.pulsar.client.restartable", "false");
+		clientFactory.setEnvironment(env);
+		assertThat(clientFactory.createClient()).isNotInstanceOf(PulsarClientProxy.class);
+	}
+
 }

spring-pulsar/src/test/java/org/springframework/pulsar/core/PulsarClientProxyTests.java

@@ -0,0 +1,204 @@
+/*
+ * 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.pulsar.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.Schema;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+
+import org.springframework.pulsar.test.support.PulsarTestContainerSupport;
+import org.springframework.test.util.ReflectionTestUtils;
+
+/**
+ * Tests for {@link PulsarClientProxy}.
+ *
+ * @author Chris Bono
+ */
+class PulsarClientProxyTests implements PulsarTestContainerSupport {
+
+	@Test
+	void constructWithCustomizer() throws Exception {
+		var restartableClient = new PulsarClientProxy(
+				(clientBuilder) -> clientBuilder.serviceUrl("pulsar://localhost:5150"));
+		restartableClient.afterPropertiesSet();
+		assertThat(restartableClient.getInstance()).hasFieldOrPropertyWithValue("conf.serviceUrl",
+				"pulsar://localhost:5150");
+	}
+
+	@Test
+	void constructWithNullCustomizer() {
+		assertThatIllegalArgumentException().isThrownBy(() -> new PulsarClientProxy(null))
+			.withMessage("customizer must not be null");
+	}
+
+	@Test
+	void restartLifecycle() throws Exception {
+		var serviceUrl = PulsarTestContainerSupport.getPulsarBrokerUrl();
+		var restartableClient = new PulsarClientProxy((builder) -> builder.serviceUrl(serviceUrl));
+		restartableClient.afterPropertiesSet();
+		var delegateClient = restartableClient.getInstance();
+		assertThat(delegateClient).isNotNull();
+		assertThat(delegateClient).hasFieldOrPropertyWithValue("conf.serviceUrl", serviceUrl);
+		assertThat(delegateClient.isClosed()).isFalse();
+		assertThat(restartableClient.isClosed()).isFalse();
+
+		// Stop and verify the client is closed
+		restartableClient.stop();
+		assertThat(restartableClient.isClosed()).isTrue();
+		assertThat(delegateClient.isClosed()).isTrue();
+		assertThat(restartableClient.getInstance()).isSameAs(delegateClient);
+
+		// Restart and verify the client is created again during start
+		restartableClient.start();
+		var newDelegateClient = restartableClient.getInstance();
+		assertThat(newDelegateClient).isNotNull();
+		assertThat(newDelegateClient).hasFieldOrPropertyWithValue("conf.serviceUrl", serviceUrl);
+		assertThat(newDelegateClient.isClosed()).isFalse();
+		assertThat(restartableClient.isClosed()).isFalse();
+		assertThat(newDelegateClient).isNotSameAs(delegateClient);
+
+		// Destroy and verify the client is destroyed as well
+		restartableClient.destroy();
+		assertThat(newDelegateClient.isClosed()).isTrue();
+		assertThat(restartableClient.getInstance()).isNull();
+	}
+
+	@Nested
+	class DelegateClientTests {
+
+		private PulsarClientProxy restartableClient;
+
+		private PulsarClient delegateClient;
+
+		@BeforeEach
+		void createClient() throws Exception {
+			var serviceUrl = PulsarTestContainerSupport.getPulsarBrokerUrl();
+			restartableClient = new PulsarClientProxy((builder) -> builder.serviceUrl(serviceUrl));
+			restartableClient.afterPropertiesSet();
+			delegateClient = mock(PulsarClient.class);
+			ReflectionTestUtils.setField(restartableClient, "instance", delegateClient);
+			assertThat(restartableClient.getInstance()).isSameAs(delegateClient);
+		}
+
+		@Test
+		void newProducer() {
+			this.restartableClient.newProducer();
+			verify(this.delegateClient).newProducer();
+		}
+
+		@Test
+		void newProducerWithSchema() {
+			this.restartableClient.newProducer(Schema.STRING);
+			verify(this.delegateClient).newProducer(Schema.STRING);
+		}
+
+		@Test
+		void newConsumer() {
+			this.restartableClient.newConsumer();
+			verify(this.delegateClient).newConsumer();
+		}
+
+		@Test
+		void newConsumerWithSchema() {
+			this.restartableClient.newConsumer(Schema.STRING);
+			verify(this.delegateClient).newConsumer(Schema.STRING);
+		}
+
+		@Test
+		void newReader() {
+			this.restartableClient.newReader();
+			verify(this.delegateClient).newReader();
+		}
+
+		@Test
+		void newReaderWithSchema() {
+			this.restartableClient.newReader(Schema.STRING);
+			verify(this.delegateClient).newReader(Schema.STRING);
+		}
+
+		@SuppressWarnings("deprecation")
+		@Test
+		void newTableViewBuilder() {
+			this.restartableClient.newTableViewBuilder(Schema.STRING);
+			verify(this.delegateClient).newTableViewBuilder(Schema.STRING);
+		}
+
+		@Test
+		void newTableView() {
+			this.restartableClient.newTableView();
+			verify(this.delegateClient).newTableView();
+		}
+
+		@Test
+		void newTableViewWithSchema() {
+			this.restartableClient.newTableView(Schema.STRING);
+			verify(this.delegateClient).newTableView(Schema.STRING);
+		}
+
+		@Test
+		void updateServiceUrl() throws PulsarClientException {
+			this.restartableClient.updateServiceUrl("pulsar://foo:6150");
+			verify(this.delegateClient).updateServiceUrl("pulsar://foo:6150");
+		}
+
+		@Test
+		void getPartitionsForTopic() {
+			this.restartableClient.getPartitionsForTopic("zTopic");
+			verify(this.delegateClient).getPartitionsForTopic("zTopic");
+		}
+
+		@Test
+		void close() throws PulsarClientException {
+			this.restartableClient.close();
+			verify(this.delegateClient).close();
+		}
+
+		@Test
+		void closeAsync() {
+			this.restartableClient.closeAsync();
+			verify(this.delegateClient).closeAsync();
+		}
+
+		@Test
+		void shutdown() throws PulsarClientException {
+			this.restartableClient.shutdown();
+			verify(this.delegateClient).shutdown();
+		}
+
+		@Test
+		void isClosed() {
+			this.restartableClient.isClosed();
+			verify(this.delegateClient).isClosed();
+		}
+
+		@Test
+		void newTransaction() {
+			this.restartableClient.newTransaction();
+			verify(this.delegateClient).newTransaction();
+		}
+
+	}
+
+}

spring-pulsar/src/test/java/org/springframework/pulsar/core/RestartableSingletonFactoryTests.java

@@ -0,0 +1,180 @@
+/*
+ * Copyright 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.pulsar.core;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+import org.junit.jupiter.params.provider.EnumSource.Mode;
+import org.mockito.InOrder;
+
+import org.springframework.pulsar.core.RestartableComponentSupport.State;
+import org.springframework.test.util.ReflectionTestUtils;
+
+/**
+ * Tests for {@link RestartableSingletonFactory}.
+ */
+public class RestartableSingletonFactoryTests {
+
+	@ParameterizedTest
+	@EnumSource(value = State.class, mode = Mode.INCLUDE, names = { "CREATED", "STOPPED" })
+	void startCreatesInstanceAndSetsStateToStarted(State initialFactoryState) {
+		var fooFactory = spy(new FooFactory());
+		setFactoryState(fooFactory, initialFactoryState);
+		assertThat(fooFactory.getInstance()).isNull();
+		fooFactory.start();
+		assertThat(fooFactory.getInstance()).isInstanceOf(Foo.class);
+		InOrder inOrder = inOrder(fooFactory);
+		inOrder.verify(fooFactory).doStart();
+		inOrder.verify(fooFactory).createInstance();
+		verifyFactoryState(fooFactory, State.STARTED);
+	}
+
+	@ParameterizedTest
+	@EnumSource(value = State.class, mode = Mode.EXCLUDE, names = { "CREATED", "STOPPED" })
+	void startDoesNothingWhenStateIsInvalid(State initialFactoryState) {
+		var fooFactory = spy(new FooFactory());
+		setFactoryState(fooFactory, initialFactoryState);
+		fooFactory.start();
+		verify(fooFactory, never()).createInstance();
+		verify(fooFactory, never()).doStart();
+	}
+
+	@ParameterizedTest
+	@EnumSource(value = State.class, mode = Mode.INCLUDE, names = { "CREATED", "STARTED" })
+	void stopStopsTheInstanceAndSetsStateToStopped(State initialFactoryState) {
+		var fooFactory = spy(new FooFactory());
+		fooFactory.start(); // creates instance we use to verify destroyInstamce
+		var foo = fooFactory.getInstance();
+		setFactoryState(fooFactory, initialFactoryState);
+		fooFactory.stop();
+		assertThat(fooFactory.getInstance()).isNull();
+		InOrder inOrder = inOrder(fooFactory);
+		inOrder.verify(fooFactory).doStop();
+		inOrder.verify(fooFactory).stopInstance(eq(foo));
+		verifyFactoryState(fooFactory, State.STOPPED);
+	}
+
+	@ParameterizedTest
+	@EnumSource(value = State.class, mode = Mode.EXCLUDE, names = { "CREATED", "STARTED" })
+	void stopDoesNothingWhenStateIsInvalid(State initialFactoryState) {
+		var fooFactory = spy(new FooFactory());
+		setFactoryState(fooFactory, initialFactoryState);
+		fooFactory.stop();
+		verify(fooFactory, never()).doStop();
+		verify(fooFactory, never()).stopInstance(any(Foo.class));
+	}
+
+	@Test
+	void stopDoesNotDiscardInstanceWhenDiscardInstanceReturnsFalse() {
+		var fooFactory = new FooFactory() {
+			@Override
+			protected boolean discardInstanceAfterStop() {
+				return false;
+			}
+		};
+		fooFactory.start();
+		fooFactory.stop();
+		assertThat(fooFactory.getInstance()).isNotNull();
+	}
+
+	@Test
+	void destroyCallsStopAndSetsStateToDestroyed() {
+		var fooFactory = spy(new FooFactory());
+		fooFactory.start();
+		fooFactory.destroy();
+		verify(fooFactory).stop();
+		verifyFactoryState(fooFactory, State.DESTROYED);
+	}
+
+	@Test
+	void initializationCreatesInstance() throws Exception {
+		var fooFactory = spy(new FooFactory());
+		fooFactory.afterPropertiesSet();
+		verify(fooFactory).createInstance();
+	}
+
+	@Test
+	void createInstanceNotCalledWhenInstanceSetInConstructor() throws Exception {
+		var foo = new Foo("the-one");
+		var fooFactory = spy(new FooFactory(foo));
+		fooFactory.afterPropertiesSet();
+		fooFactory.start();
+		verify(fooFactory, never()).createInstance();
+		assertThat(fooFactory.getInstance()).isEqualTo(foo);
+	}
+
+	@Test
+	void createInstanceOnlyCalledOnceDuringStartup() throws Exception {
+		var fooFactory = spy(new FooFactory());
+		fooFactory.afterPropertiesSet();
+		fooFactory.start();
+		verify(fooFactory).createInstance();
+	}
+
+	void setFactoryState(RestartableSingletonFactory<?> factory, State factoryState) {
+		ReflectionTestUtils.setField(factory, "state", new AtomicReference<>(factoryState));
+		verifyFactoryState(factory, factoryState);
+	}
+
+	void verifyFactoryState(RestartableSingletonFactory<?> factory, State expectedState) {
+		assertThat(factory).extracting("state").satisfies((state) -> {
+			assertThat(state.toString()).contains(expectedState.name());
+		});
+	}
+
+	static class FooFactory extends RestartableSingletonFactory<Foo> {
+
+		FooFactory() {
+			super();
+		}
+
+		FooFactory(Foo instance) {
+			super(instance);
+		}
+
+		@Override
+		protected Foo createInstance() {
+			return new Foo("restart:" + System.currentTimeMillis());
+		}
+
+	}
+
+	static class FooKeepsInstanceFactory extends FooFactory {
+
+		@Override
+		protected boolean discardInstanceAfterStop() {
+			return false;
+		}
+
+	}
+
+	record Foo(String name) {
+
+	}
+
+}