From e366b8365e6b5b60dfb5a0f48857134b239d9e13 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Thu, 12 Sep 2024 14:14:28 -0500 Subject: [PATCH] Use unchecked exception on reader factory (#845) This commit replaces the checked `PulsarClientException` with the unchecked `PulsarException` on the `PulsarReaderFactory#createReader` API. --- .../main/antora/modules/ROOT/pages/whats-new.adoc | 3 +++ .../pulsar/core/DefaultPulsarReaderFactory.java | 12 +++++++++--- .../pulsar/core/PulsarReaderFactory.java | 11 +++++++---- .../reader/DefaultPulsarMessageReaderContainer.java | 12 +++--------- .../pulsar/core/DefaultPulsarReaderFactoryTests.java | 5 +++-- 5 files changed, 25 insertions(+), 18 deletions(-) diff --git a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc index 39fb5cf4..f568efbe 100644 --- a/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc +++ b/spring-pulsar-docs/src/main/antora/modules/ROOT/pages/whats-new.adoc @@ -67,6 +67,9 @@ The `PulsarTopic` constructor now requires a fully qualified topic name (`domain If you are invoking the constructor you will need to be sure the topic you pass in is fully-qualified. A better alternative is to instead use the `PulsarTopicBuilder` as it does not require fully qualified names and will add default values for the missing components in the specified name. +==== PulsarReaderFactory#createReader +The `PulsarReaderFactory#createReader` API now throws an unchecked `PulsarException` rather than a checked `PulsarClientException`. +Replace any `try/catch` blocks on this API accordingly. [[what-s-new-in-1-1-since-1-0]] == What's New in 1.1 Since 1.0 diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java index 44a8764c..f22774b9 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/DefaultPulsarReaderFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2023 the original author or authors. + * Copyright 2023-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. @@ -30,6 +30,7 @@ import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.client.impl.ReaderBuilderImpl; import org.springframework.lang.Nullable; +import org.springframework.pulsar.PulsarException; import org.springframework.util.CollectionUtils; /** @@ -82,7 +83,7 @@ public class DefaultPulsarReaderFactory implements PulsarReaderFactory { @Override public Reader createReader(@Nullable List topics, @Nullable MessageId messageId, Schema schema, - @Nullable List> customizers) throws PulsarClientException { + @Nullable List> customizers) { Objects.requireNonNull(schema, "Schema must be specified"); ReaderBuilder readerBuilder = this.pulsarClient.newReader(schema); @@ -103,7 +104,12 @@ public class DefaultPulsarReaderFactory implements PulsarReaderFactory { customizers.forEach(customizer -> customizer.customize(readerBuilder)); } - return readerBuilder.create(); + try { + return readerBuilder.create(); + } + catch (PulsarClientException ex) { + throw new PulsarException(ex); + } } private void replaceTopicsOnBuilder(ReaderBuilder builder, Collection topics) { diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarReaderFactory.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarReaderFactory.java index 50a01020..5ad0dbbc 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarReaderFactory.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarReaderFactory.java @@ -1,5 +1,5 @@ /* - * Copyright 2023 the original author or authors. + * Copyright 2023-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. @@ -24,12 +24,14 @@ import org.apache.pulsar.client.api.Reader; import org.apache.pulsar.client.api.Schema; import org.springframework.lang.Nullable; +import org.springframework.pulsar.PulsarException; /** * Pulsar {@link Reader} factory interface. * * @param Underlying message type handled by this reader. * @author Soby Chacko + * @author Chris Bono */ public interface PulsarReaderFactory { @@ -42,10 +44,11 @@ public interface PulsarReaderFactory { * @param customizers the optional list of customizers to apply to the reader builder. * Note that the customizers are applied last and have the potential for overriding * any specified parameters or default properties. - * @return Pulsar {@link Reader} - * @throws PulsarClientException if there are issues when creating the reader + * @return the created reader + * @throws PulsarException if any {@link PulsarClientException} occurs communicating + * with Pulsar */ Reader createReader(@Nullable List topics, @Nullable MessageId messageId, Schema schema, - @Nullable List> customizers) throws PulsarClientException; + @Nullable List> customizers); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainer.java b/spring-pulsar/src/main/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainer.java index b8287f29..7001978d 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainer.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/reader/DefaultPulsarMessageReaderContainer.java @@ -192,15 +192,9 @@ public class DefaultPulsarMessageReaderContainer extends AbstractPulsarMessag this.readerBuilderCustomizer = getReaderBuilderCustomizer(); List> customizers = this.readerBuilderCustomizer != null ? List.of(this.readerBuilderCustomizer) : Collections.emptyList(); - try { - this.reader = getPulsarReaderFactory().createReader(readerContainerProperties.getTopics(), - readerContainerProperties.getStartMessageId(), (Schema) readerContainerProperties.getSchema(), - customizers); - } - catch (PulsarClientException ex) { - // TODO remove when PRF.createReader replaces PCEX w PEX - throw new PulsarException(ex); - } + this.reader = getPulsarReaderFactory().createReader(readerContainerProperties.getTopics(), + readerContainerProperties.getStartMessageId(), (Schema) readerContainerProperties.getSchema(), + customizers); } @Override diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java index 1afec2f4..c7cb6c41 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/core/DefaultPulsarReaderFactoryTests.java @@ -41,6 +41,7 @@ import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.mockito.InOrder; +import org.springframework.pulsar.PulsarException; import org.springframework.pulsar.test.support.PulsarTestContainerSupport; /** @@ -272,7 +273,7 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo // topic name is not set in the API call or in the reader config. assertThatThrownBy(() -> pulsarReaderFactory.createReader(Collections.emptyList(), MessageId.earliest, Schema.STRING, Collections.emptyList())) - .isInstanceOf(PulsarClientException.class) + .isInstanceOf(PulsarException.class) .hasMessageContaining("Topic name must be set on the reader builder"); } @@ -280,7 +281,7 @@ public class DefaultPulsarReaderFactoryTests implements PulsarTestContainerSuppo void missingStartingMessageId() { assertThatThrownBy(() -> pulsarReaderFactory.createReader(List.of("my-reader-topic"), null, Schema.STRING, Collections.emptyList())) - .isInstanceOf(PulsarClientException.class) + .isInstanceOf(PulsarException.class) .hasMessageContaining( "Start message id or start message from roll back must be specified but they cannot be specified at the same time"); }