From ea1597a0a1e810f5e9a924f34b5dc3303f9d2243 Mon Sep 17 00:00:00 2001 From: Soby Chacko Date: Fri, 20 Jan 2023 16:21:38 -0500 Subject: [PATCH] Pulsar binder subscription name improvements Provide a default subscription name that is anonymous each time the app starts. This can be overridden to a fixed subscription name by providing it as a property. --- .../src/main/resources/application.yml | 1 - .../stream/binder/PulsarBinderUtils.java | 51 +++++++++++++++++++ .../binder/PulsarMessageChannelBinder.java | 3 +- .../properties/PulsarConsumerProperties.java | 6 +-- .../stream/binder/PulsarBinderUtilsTests.java | 50 ++++++++++++++++++ 5 files changed, 106 insertions(+), 5 deletions(-) create mode 100644 spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java create mode 100644 spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java diff --git a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml b/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml index 106ff7cd..b460aeb2 100644 --- a/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml +++ b/spring-pulsar-sample-apps/sample-pulsar-binder/src/main/resources/application.yml @@ -15,7 +15,6 @@ spring: bindings: timeLogger-in-0: consumer: - subscription-name: my-scst-sub1 schema-type: STRING timeSupplier-out-0: producer: diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java new file mode 100644 index 00000000..25013e00 --- /dev/null +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtils.java @@ -0,0 +1,51 @@ +/* + * 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.spring.cloud.stream.binder; + +import java.util.UUID; + +import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties; + +/** + * Utility methods for the binder. + * + * @author Soby Chacko + */ +public final class PulsarBinderUtils { + + static final String ANON_SUBSCRIPTION = "anon-subscription"; + + static final char SUBSCRIPTION_NAME_SEPARATOR = '-'; + + private PulsarBinderUtils() { + + } + + static String subscriptionName(PulsarConsumerProperties pulsarConsumerProperties, + ConsumerDestination consumerDestination) { + String subscriptionName = pulsarConsumerProperties.getSubscriptionName(); + if (subscriptionName == null) { + // if subscription name is not provided, each time the app starts, it will be + // an anonymous subscription + subscriptionName = consumerDestination.getName() + SUBSCRIPTION_NAME_SEPARATOR + ANON_SUBSCRIPTION + + SUBSCRIPTION_NAME_SEPARATOR + UUID.randomUUID(); + } + return subscriptionName; + } + +} diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java index 10f9c1c4..9564f7f1 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarMessageChannelBinder.java @@ -110,7 +110,8 @@ public class PulsarMessageChannelBinder extends else { pulsarContainerProperties.setSchema(Schema.BYTES); } - pulsarContainerProperties.setSubscriptionName(properties.getExtension().getSubscriptionName()); + String subscriptionName = PulsarBinderUtils.subscriptionName(properties.getExtension(), destination); + pulsarContainerProperties.setSubscriptionName(subscriptionName); DefaultPulsarMessageListenerContainer container = new DefaultPulsarMessageListenerContainer<>( this.pulsarConsumerFactory, pulsarContainerProperties); pulsarMessageDrivenChannelAdapter.setMessageListenerContainer(container); diff --git a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java index 429121ce..5bf9938c 100644 --- a/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java +++ b/spring-pulsar-spring-cloud-stream-binder/src/main/java/org/springframework/pulsar/spring/cloud/stream/binder/properties/PulsarConsumerProperties.java @@ -19,7 +19,6 @@ package org.springframework.pulsar.spring.cloud.stream.binder.properties; import org.apache.pulsar.client.api.SubscriptionType; import org.apache.pulsar.common.schema.SchemaType; -import org.springframework.lang.NonNull; import org.springframework.lang.Nullable; /** @@ -29,8 +28,8 @@ import org.springframework.lang.Nullable; */ public class PulsarConsumerProperties { - @NonNull - private String subscriptionName = "DEFAULT-SUBSCRIPTION-TODO-CHANGE-THIS"; + @Nullable + private String subscriptionName; @Nullable private SchemaType schemaType; @@ -38,6 +37,7 @@ public class PulsarConsumerProperties { @Nullable private SubscriptionType subscriptionType; + @Nullable public String getSubscriptionName() { return this.subscriptionName; } diff --git a/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java new file mode 100644 index 00000000..8112bf29 --- /dev/null +++ b/spring-pulsar-spring-cloud-stream-binder/src/test/java/org/springframework/pulsar/spring/cloud/stream/binder/PulsarBinderUtilsTests.java @@ -0,0 +1,50 @@ +/* + * 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.spring.cloud.stream.binder; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +import org.junit.jupiter.api.Test; + +import org.springframework.cloud.stream.provisioning.ConsumerDestination; +import org.springframework.pulsar.spring.cloud.stream.binder.properties.PulsarConsumerProperties; + +public class PulsarBinderUtilsTests { + + @Test + void subscriptionNameIsNotNullWhenProvidedAsProperty() { + ConsumerDestination consumerDestination = mock(ConsumerDestination.class); + PulsarConsumerProperties pulsarConsumerProperties = mock(PulsarConsumerProperties.class); + when(pulsarConsumerProperties.getSubscriptionName()).thenReturn("my-subscription"); + String subscriptionName = PulsarBinderUtils.subscriptionName(pulsarConsumerProperties, consumerDestination); + assertThat(subscriptionName).isEqualTo("my-subscription"); + } + + @Test + void subscriptionNameIsNotNullWhenPropertyIsMissing() { + ConsumerDestination consumerDestination = mock(ConsumerDestination.class); + PulsarConsumerProperties pulsarConsumerProperties = mock(PulsarConsumerProperties.class); + when(pulsarConsumerProperties.getSubscriptionName()).thenReturn(null); + when(consumerDestination.getName()).thenReturn("my-topic"); + String subscriptionName = PulsarBinderUtils.subscriptionName(pulsarConsumerProperties, consumerDestination); + assertThat(subscriptionName).startsWith("my-topic" + PulsarBinderUtils.SUBSCRIPTION_NAME_SEPARATOR + + PulsarBinderUtils.ANON_SUBSCRIPTION + PulsarBinderUtils.SUBSCRIPTION_NAME_SEPARATOR); + } + +}