Ignore exceptions when closing consumers, producers and readers (#218)

Author: lhotari

pulsar-client-reactive-adapter/src/main/java/org/apache/pulsar/reactive/client/internal/adapter/ReactiveConsumerAdapter.java

@@ -54,12 +54,17 @@ private Mono<Consumer<T>> createConsumerMono() {
 	}
 
 	private Mono<Void> closeConsumer(Consumer<?> consumer) {
-		return Mono.deferContextual((contextView) -> Mono.fromFuture(consumer::closeAsync).doFinally((signalType) -> {
-			this.LOG.info("Consumer closed {}", consumer);
-			contextView.<InternalConsumerListener>getOrEmpty(InternalConsumerListener.class)
-				.ifPresent((listener) -> listener.onConsumerClosed(consumer));
-		}));
-
+		return Mono
+			.deferContextual((contextView) -> AdapterImplementationFactory.adaptPulsarFuture(consumer::closeAsync)
+				.onErrorResume((t) -> {
+					this.LOG.debug("Error closing consumer {}", consumer, t);
+					return Mono.empty();
+				})
+				.doFinally((signalType) -> {
+					this.LOG.info("Consumer closed {}", consumer);
+					contextView.<InternalConsumerListener>getOrEmpty(InternalConsumerListener.class)
+						.ifPresent((listener) -> listener.onConsumerClosed(consumer));
+				}));
 	}
 
 	<R> Mono<R> usingConsumer(Function<Consumer<T>, Mono<R>> usingConsumerAction) {

pulsar-client-reactive-adapter/src/main/java/org/apache/pulsar/reactive/client/internal/adapter/ReactiveProducerAdapter.java

@@ -28,6 +28,8 @@
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.impl.ProducerBuilderImpl;
 import org.apache.pulsar.reactive.client.internal.api.PublisherTransformer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.Disposable;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -36,6 +38,8 @@
 
 class ReactiveProducerAdapter<T> {
 
+	private static final Logger log = LoggerFactory.getLogger(ReactiveProducerAdapter.class);
+
 	private final ProducerCache producerCache;
 
 	private final Function<PulsarClient, ProducerBuilder<T>> producerBuilderFactory;
@@ -73,7 +77,10 @@ private Mono<Tuple2<ProducerCacheKey, Mono<Producer<T>>>> createCachedProducerKe
 	}
 
 	private Mono<Void> closeProducer(Producer<?> producer) {
-		return AdapterImplementationFactory.adaptPulsarFuture(producer::closeAsync);
+		return AdapterImplementationFactory.adaptPulsarFuture(producer::closeAsync).onErrorResume((t) -> {
+			log.debug("Error closing producer {}", producer, t);
+			return Mono.empty();
+		});
 	}
 
 	<R> Mono<R> usingProducer(BiFunction<Producer<T>, PublisherTransformer, Mono<R>> usingProducerAction) {

pulsar-client-reactive-adapter/src/main/java/org/apache/pulsar/reactive/client/internal/adapter/ReactiveReaderAdapter.java

@@ -25,11 +25,15 @@
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.Reader;
 import org.apache.pulsar.client.api.ReaderBuilder;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 
 class ReactiveReaderAdapter<T> {
 
+	private static final Logger log = LoggerFactory.getLogger(ReactiveReaderAdapter.class);
+
 	private final Supplier<PulsarClient> pulsarClientSupplier;
 
 	private final Function<PulsarClient, ReaderBuilder<T>> readerBuilderFactory;
@@ -46,7 +50,10 @@ private Mono<Reader<T>> createReaderMono() {
 	}
 
 	private Mono<Void> closeReader(Reader<?> reader) {
-		return AdapterImplementationFactory.adaptPulsarFuture(reader::closeAsync);
+		return AdapterImplementationFactory.adaptPulsarFuture(reader::closeAsync).onErrorResume((t) -> {
+			log.debug("Error closing reader {}", reader, t);
+			return Mono.empty();
+		});
 	}
 
 	<R> Mono<R> usingReader(Function<Reader<T>, Mono<R>> usingReaderAction) {

pulsar-client-reactive-adapter/src/test/java/org/apache/pulsar/reactive/client/internal/adapter/AdaptedReactiveMessageConsumerTests.java

@@ -40,6 +40,7 @@
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
 import org.apache.pulsar.client.api.RegexSubscriptionMode;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.api.SubscriptionInitialPosition;
@@ -406,4 +407,26 @@ void consumePulsarException() throws Exception {
 			.verifyError(PulsarClientException.InvalidMessageException.class);
 	}
 
+	@Test
+	void closeConsumerExceptionIsIgnored() throws Exception {
+		PulsarClientImpl pulsarClient = spy(
+				(PulsarClientImpl) PulsarClient.builder().serviceUrl("http://dummy").build());
+
+		Consumer<String> consumer = mock(Consumer.class);
+		doReturn(CompletableFuture.failedFuture(new AlreadyClosedException("Already closed"))).when(consumer)
+			.closeAsync();
+
+		doReturn(CompletableFuture.completedFuture(consumer)).when(pulsarClient)
+			.subscribeAsync(any(ConsumerConfigurationData.class), eq(Schema.STRING), isNull());
+
+		ReactiveMessageConsumer<String> reactiveConsumer = AdaptedReactivePulsarClientFactory.create(pulsarClient)
+			.messageConsumer(Schema.STRING)
+			.topic("my-topic")
+			.subscriptionName("my-sub")
+			.build();
+
+		StepVerifier.create(reactiveConsumer.consumeNothing()).expectComplete().verify();
+		verify(consumer).closeAsync();
+	}
+
 }

pulsar-client-reactive-adapter/src/test/java/org/apache/pulsar/reactive/client/internal/adapter/AdaptedReactiveMessageReaderTests.java

@@ -30,6 +30,7 @@
 import org.apache.pulsar.client.api.MessageId;
 import org.apache.pulsar.client.api.PulsarClient;
 import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
 import org.apache.pulsar.client.api.Range;
 import org.apache.pulsar.client.api.Reader;
 import org.apache.pulsar.client.api.Schema;
@@ -168,6 +169,29 @@ void readPulsarException() throws Exception {
 		StepVerifier.create(reactiveReader.readMany()).verifyError(PulsarClientException.InvalidMessageException.class);
 	}
 
+	@Test
+	void closeReaderExceptionIsIgnored() throws Exception {
+		PulsarClientImpl pulsarClient = spy(
+				(PulsarClientImpl) PulsarClient.builder().serviceUrl("http://dummy").build());
+
+		Reader<String> reader = mock(Reader.class);
+		doReturn(CompletableFuture.failedFuture(new AlreadyClosedException("Already closed"))).when(reader)
+			.closeAsync();
+		doReturn(CompletableFuture.completedFuture(false)).when(reader).hasMessageAvailableAsync();
+
+		doReturn(CompletableFuture.completedFuture(reader)).when(pulsarClient)
+			.createReaderAsync(any(), eq(Schema.STRING));
+
+		ReactiveMessageReader<String> reactiveReader = AdaptedReactivePulsarClientFactory.create(pulsarClient)
+			.messageReader(Schema.STRING)
+			.endOfStreamAction(EndOfStreamAction.COMPLETE)
+			.topic("my-topic")
+			.build();
+
+		StepVerifier.create(reactiveReader.readOne()).expectComplete().verify();
+		verify(reader).closeAsync();
+	}
+
 	@Test
 	void endOfStreamPoll() throws Exception {
 		PulsarClientImpl pulsarClient = spy(

pulsar-client-reactive-adapter/src/test/java/org/apache/pulsar/reactive/client/internal/adapter/AdaptedReactiveMessageSenderTests.java

@@ -47,6 +47,7 @@
 import org.apache.pulsar.client.api.ProducerAccessMode;
 import org.apache.pulsar.client.api.ProducerCryptoFailureAction;
 import org.apache.pulsar.client.api.PulsarClient;
+import org.apache.pulsar.client.api.PulsarClientException.AlreadyClosedException;
 import org.apache.pulsar.client.api.PulsarClientException.ProducerQueueIsFullError;
 import org.apache.pulsar.client.api.Schema;
 import org.apache.pulsar.client.impl.MessageIdImpl;
@@ -547,4 +548,35 @@ void doTestMaxInFlight(BiFunction<ReactiveMessageSender<String>, Flux<Integer>,
 		}
 	}
 
+	@Test
+	void closeProducerExceptionIsIgnored() throws Exception {
+		PulsarClientImpl pulsarClient = spy(
+				(PulsarClientImpl) PulsarClient.builder().serviceUrl("http://dummy").build());
+
+		ProducerBase<String> producer = mock(ProducerBase.class);
+		doReturn(CompletableFuture.failedFuture(new AlreadyClosedException("Already closed"))).when(producer)
+			.closeAsync();
+
+		MessageId messageId = DefaultImplementation.getDefaultImplementation().newMessageId(1, 1, 1);
+
+		given(producer.newMessage()).willAnswer((__) -> {
+			TypedMessageBuilderImpl<String> typedMessageBuilder = spy(
+					new TypedMessageBuilderImpl<>(producer, Schema.STRING));
+			willAnswer((___) -> CompletableFuture.completedFuture(messageId)).given(typedMessageBuilder).sendAsync();
+			return typedMessageBuilder;
+		});
+
+		doReturn(CompletableFuture.completedFuture(producer)).when(pulsarClient)
+			.createProducerAsync(any(), eq(Schema.STRING), isNull());
+
+		ReactiveMessageSender<String> reactiveSender = AdaptedReactivePulsarClientFactory.create(pulsarClient)
+			.messageSender(Schema.STRING)
+			.topic("my-topic")
+			.build();
+
+		StepVerifier.create(reactiveSender.sendOne(MessageSpec.of("test1"))).expectNext(messageId).verifyComplete();
+
+		verify(producer).closeAsync();
+	}
+
 }