diff --git a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java index b036c008..81d12a08 100644 --- a/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java +++ b/buildSrc/src/main/java/org/springframework/pulsar/gradle/docs/configprops/DocumentConfigurationProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2012-2022 the original author or authors. + * Copyright 2012-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. @@ -73,17 +73,14 @@ public class DocumentConfigurationProperties extends DefaultTask { c.accept("spring.pulsar.consumer"); c.accept("spring.pulsar.listener"); }); + snippets.add("application-properties.pulsar-function", "Pulsar Function Properties", (c) -> c.accept("spring.pulsar.function")); snippets.add("application-properties.pulsar-administration", "Pulsar Administration Properties", (c) -> c.accept("spring.pulsar.administration")); - snippets.add("application-properties.pulsar-reactive-sender", "Pulsar Reactive Sender Properties", (c) -> { - c.accept("spring.pulsar.reactive.sender"); - }); + snippets.add("application-properties.pulsar-reactive-sender", "Pulsar Reactive Sender Properties", (c) -> c.accept("spring.pulsar.reactive.sender")); snippets.add("application-properties.pulsar-reactive-consumer", "Pulsar Reactive Consumer Properties", (c) -> { c.accept("spring.pulsar.reactive.consumer"); c.accept("spring.pulsar.reactive.listener"); }); - snippets.add("application-properties.pulsar-reactive-reader", "Pulsar Reactive Reader Properties", (c) -> { - c.accept("spring.pulsar.reactive.reader"); - }); + snippets.add("application-properties.pulsar-reactive-reader", "Pulsar Reactive Reader Properties", (c) -> c.accept("spring.pulsar.reactive.reader")); snippets.writeTo(this.outputDir.toPath()); } } diff --git a/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc b/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc index 86c0f107..825a1d31 100644 --- a/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/application-properties.adoc @@ -18,6 +18,8 @@ include::application-properties/pulsar-producer.adoc[] include::application-properties/pulsar-consumer.adoc[] +include::application-properties/pulsar-function.adoc[] + include::application-properties/pulsar-administration.adoc[] include::application-properties/pulsar-reactive-sender.adoc[] diff --git a/spring-pulsar-docs/src/main/asciidoc/attributes.adoc b/spring-pulsar-docs/src/main/asciidoc/attributes.adoc index 9235b2ea..338d9443 100644 --- a/spring-pulsar-docs/src/main/asciidoc/attributes.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/attributes.adoc @@ -12,3 +12,7 @@ :github: https://github.com/spring-projects-experimental/spring-pulsar :javadocs: https://docs.spring.io/spring-pulsar/docs/{spring-pulsar-version}/api :spring-boot-docs: https://docs.spring.io/spring-boot/docs/{spring-boot-version}/reference/htmlsingle + +:apache-pulsar-docs: https://pulsar.apache.org/docs/2.10.x +:apache-pulsar-io-docs: {apache-pulsar-docs}/io-connectors +:apache-pulsar-function-docs: {apache-pulsar-docs}/functions-overview diff --git a/spring-pulsar-docs/src/main/asciidoc/copyright.adoc b/spring-pulsar-docs/src/main/asciidoc/copyright.adoc index be2f6b50..ba868fe0 100644 --- a/spring-pulsar-docs/src/main/asciidoc/copyright.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/copyright.adoc @@ -1,3 +1,3 @@ -(C) 2022 VMware, Inc. +(C) 2022-2023 VMware, Inc. Copies of this document may be made for your own use and for distribution to others, provided that you do not charge any fee for such copies and further provided that each copy contains this Copyright Notice, whether distributed in print or electronically. diff --git a/spring-pulsar-docs/src/main/asciidoc/index.adoc b/spring-pulsar-docs/src/main/asciidoc/index.adoc index eab80b7b..4480ce71 100644 --- a/spring-pulsar-docs/src/main/asciidoc/index.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/index.adoc @@ -21,6 +21,8 @@ include::reactive-pulsar.adoc[leveloffset=+2] include::pulsar-admin.adoc[leveloffset=+2] +include::pulsar-function.adoc[leveloffset=+2] + include::observability.adoc[leveloffset=+2] :sectnums!: diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar-admin.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar-admin.adoc index 99ac69a8..ffcd2ed7 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar-admin.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar-admin.adoc @@ -1,8 +1,8 @@ [[pulsar-admin]] = Pulsar Administration +include::attributes.adoc[] -:javadocs: https://docs.spring.io/spring-pulsar/docs/current-SNAPSHOT/api - +[[pulsar-admin-client]] == Pulsar Admin Client On the Pulsar administration side, Spring Boot auto-configuration provides a `PulsarAdministration` to manage Pulsar clusters. The administration implements an interface called `PulsarAdminOperations` and provides {javadocs}/org/springframework/pulsar/core/PulsarAdminOperations.html[a `createOrModify` method] to handle topic administration through its contract. diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc new file mode 100644 index 00000000..16564365 --- /dev/null +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc @@ -0,0 +1,128 @@ +[[pulsar-function]] += Pulsar Functions +include::attributes.adoc[] + +Spring for Apache Pulsar provides basic suppport for {apache-pulsar-io-docs}/[Pulsar IO] (connectors) and {apache-pulsar-function-docs}[Pulsar Functions] which allow users to define stream processing pipelines made up of `sources`, `processors`, and `sinks`. +The `sources` and `sinks` are modeled by _Pulsar IO (connectors)_ and the `processors` are represented by _Pulsar Functions_. + +NOTE: Because connectors are just special functions, and for simplicity, we refer to sources, sinks and functions collectively as "Pulsar Functions". + +.Pre-requisites +**** + +*Familiarity* - the audience is expected to be somewhat familiar w/ _Pulsar IO_ and _Pulsar Functions_. +If that is not the case it may be helpful to see their getting started guides. + +*Feature enabled* - to use these features the functions support in Apache Pulsar must be enabled and configured (it is disabled by default). +The built-in connectors may also need to be installed on the Pulsar cluster. + +See the {apache-pulsar-io-docs}/[Pulsar IO] and {apache-pulsar-function-docs}[Pulsar Functions] docs for more details. +**** + +== Pulsar Function Administration +The framework provides the `PulsarFunctionAdministration` component to manage Pulsar functions. +When you use the Pulsar Spring Boot starter, you get the `PulsarFunctionAdministration` auto-configured. + +By default, the application tries to connect to a local Pulsar instance at `http://localhost:8080`. +However, because it leverages the already configured `PulsarAdministration`, see <> for available client options (including authentication). Other available application properties can be found in the <> prefixed by `spring.pulsar.function`. + +== Automatic Function Management +On initialization, the framework finds all `PulsarFunction`, `PulsarSink`, and `PulsarSource` beans in the application context. +For each bean, the corresponding Pulsar function is either created or updated. +The proper API is called based on function type, function config, and whether the function already exists. + +NOTE: The `PulsarFunction`, `PulsarSink`, and `PulsarSource` beans are simple wrappers around the Apache Pulsar config objects `FunctionConfig`, `SinkConfig`, and `SourceConfig`, respectively. +Due to the large number of supported connectors (and their varied configurations) the framework does not attempt to create a configuration properties hierarchy to mirror the varied Apache Pulsar connectors. +Instead, the burden is on the user to supply the full config object and then the framework handles the management (create/update) using the supplied config. + +== Limitations + +=== No Magic Pulsar Functions +Pulsar functions and custom connectors are represented by custom application code (eg. a `java.util.Function`). +There is no magic support to automatically register the custom code. +While this would be amazing, it has some technical challenges and not yet been implemented. +As such, it is up to the user to ensure the function (or custom connector) is available at the location specified in the function config. +For example, if the function config has a `jar` value of `./some/path/MyFunction.jar` then the function jar file must exist at the specified path. + +=== Name Identifier +The `name` property from the function config is used as the identifier to determine if a function already exists in order to decide if an update or create operation is performed. +As such, the name should not be modified if function updates are desired. + +== Configuration + +=== Pulsar Function Archive +Each Pulsar function is represented by an actual archive (eg. jar file). +The path to the archive is specified via the `archive` property for sources and sinks, and the `jar` property for functions. + +The following rules determine the "type" of path: + +* The path is a *URL* when it starts w/ `(file|http|https|function|sink|source)://` +* The path is *built-in* when it starts w/ `builtin://` (points to one of the provided out-of-the-box connectors) +* The path is *local* otherwise. + +The action that occurs during the create/update operation is dependent on path "type" as follows: + +* When the path is a *URL* the content is downloaded by the server +* When the path is *built-in* the content is already available on the server +* When the path is *local* the content is uploaded to the server + +=== Built-in Source and Sinks +Apache Pulsar provides many source and sink connectors out-of-the-box, aka built-in connectors. To use a built-in connector simply set the `archive` to `builtin://` (eg `builtin://rabbit`). + +== Examples + +Here are some examples that show how to configure a `PulsarSource` bean which results in the `PulsarFunctionAdministration` auto-creating the backing Pulsar source connector. + +.PulsarSource using built-in Rabbit connector +==== +[source,java] +---- +@Bean +PulsarSource rabbitSource() { + Map configs = new HashMap<>(); + configs.put("host", "my.rabbit.host"); + configs.put("port", 5672); + configs.put("virtualHost", "/"); + configs.put("username", "guest"); + configs.put("password", "guest"); + configs.put("queueName", "test_rabbit"); + configs.put("connectionName", "test-connection"); + SourceConfig sourceConfig = SourceConfig.builder() + .tenant("public") + .namespace("default") + .name("rabbit-test-source") + .archive("builtin://rabbitmq") + .topicName("incoming_rabbit") + .configs(configs).build(); + return new PulsarSource(sourceConfig, null); +} +---- +==== + +This next example is the same as the previous, except that it uses the Spring Boot auto-configured `RabbitProperties` to ease the configuration burden. This of course requires the application to be using Spring Boot with Rabbit auto-configuration enabled. + +.PulsarSource using built-in Rabbit connector and Spring Boot RabbitProperties +==== +[source,java] +---- +@Bean +PulsarSource rabbitSourceWithBootProps(RabbitProperties props) { + Map configs = new HashMap<>(); + configs.put("host", props.determineHost()); + configs.put("port", props.determinePort()); + configs.put("virtualHost", props.determineVirtualHost()); + configs.put("username", props.determineUsername()); + configs.put("password", props.determinePassword()); + configs.put("queueName", "test_rabbit"); + configs.put("connectionName", "test-connection"); + SourceConfig sourceConfig = SourceConfig.builder() + .tenant("public") + .namespace("default") + .name("rabbit-test-source") + .archive("builtin://rabbitmq") + .topicName("incoming_rabbit") + .configs(configs).build(); + return new PulsarSource(sourceConfig, null); +} +---- +==== diff --git a/spring-pulsar-spring-boot-autoconfigure/build.gradle b/spring-pulsar-spring-boot-autoconfigure/build.gradle index ddbc74da..207972e7 100644 --- a/spring-pulsar-spring-boot-autoconfigure/build.gradle +++ b/spring-pulsar-spring-boot-autoconfigure/build.gradle @@ -21,6 +21,8 @@ dependencies { testImplementation 'org.springframework.boot:spring-boot-starter-web' testImplementation 'org.testcontainers:junit-jupiter' testImplementation 'org.testcontainers:pulsar' + // used by PulsarFunctionTests + testImplementation 'org.testcontainers:rabbitmq' } test { diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java index 5e4bb1d8..fe43cb59 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * 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. @@ -37,6 +37,10 @@ import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.function.PulsarFunction; +import org.springframework.pulsar.function.PulsarFunctionAdministration; +import org.springframework.pulsar.function.PulsarSink; +import org.springframework.pulsar.function.PulsarSource; import org.springframework.pulsar.observation.PulsarTemplateObservationConvention; import io.micrometer.observation.ObservationRegistry; @@ -113,4 +117,14 @@ public class PulsarAutoConfiguration { return new PulsarAdministration(this.properties.buildAdminProperties()); } + @Bean + @ConditionalOnMissingBean(PulsarFunctionAdministration.class) + @ConditionalOnProperty(name = "spring.pulsar.function.enabled", havingValue = "true", matchIfMissing = true) + public PulsarFunctionAdministration pulsarFunctionAdministration(PulsarAdministration pulsarAdministration, + ObjectProvider pulsarFunctions, ObjectProvider pulsarSinks, + ObjectProvider pulsarSources) { + return new PulsarFunctionAdministration(pulsarAdministration, pulsarFunctions, pulsarSinks, pulsarSources, + this.properties.getFunction().getFailFast(), this.properties.getFunction().getPropagateFailures()); + } + } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java index e2661174..8e3493a4 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * 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. @@ -67,6 +67,8 @@ public class PulsarProperties { private final Client client = new Client(); + private final Function function = new Function(); + private final Listener listener = new Listener(); private final Producer producer = new Producer(); @@ -87,6 +89,10 @@ public class PulsarProperties { return this.listener; } + public Function getFunction() { + return this.function; + } + public Producer getProducer() { return this.producer; } @@ -733,7 +739,7 @@ public class PulsarProperties { */ private Map properties = new HashMap<>(); - private Cache cache = new Cache(); + private final Cache cache = new Cache(); public String getTopicName() { return this.topicName; @@ -1654,6 +1660,37 @@ public class PulsarProperties { } + public static class Function { + + /** + * Whether to stop processing further function creates/updates when a failure + * occurs. + */ + private Boolean failFast = Boolean.TRUE; + + /** + * Whether to throw an exception if any failure is encountered during processing. + */ + private Boolean propagateFailures = Boolean.TRUE; + + public Boolean getFailFast() { + return this.failFast; + } + + public void setFailFast(Boolean failFast) { + this.failFast = failFast; + } + + public Boolean getPropagateFailures() { + return this.propagateFailures; + } + + public void setPropagateFailures(Boolean propagateFailures) { + this.propagateFailures = propagateFailures; + } + + } + public static class Listener { /** @@ -2015,7 +2052,6 @@ public class PulsarProperties { } - @SuppressWarnings("serial") private static class Properties extends HashMap { java.util.function.Consumer in(String key) { diff --git a/spring-pulsar-spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json b/spring-pulsar-spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json new file mode 100644 index 00000000..6fe034ef --- /dev/null +++ b/spring-pulsar-spring-boot-autoconfigure/src/main/resources/META-INF/additional-spring-configuration-metadata.json @@ -0,0 +1,24 @@ +{ + "groups": [], + "properties": [ + { + "name": "spring.pulsar.function.enabled", + "type": "java.lang.Boolean", + "description": "Whether to enable function support.", + "defaultValue": true + }, + { + "name": "spring.pulsar.producer.cache.enabled", + "type": "java.lang.Boolean", + "description": "Whether to enable caching in the PulsarProducerFactory.", + "defaultValue": true + }, + { + "name": "spring.pulsar.reactive.sender.cache.enabled", + "type": "java.lang.Boolean", + "description": "Whether to enable caching in the ReactivePulsarSenderFactory.", + "defaultValue": true + } + ], + "hints": [] +} diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java index 8658f49b..ff38b510 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * 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. @@ -51,6 +51,7 @@ import org.springframework.pulsar.core.PulsarAdministration; import org.springframework.pulsar.core.PulsarConsumerFactory; import org.springframework.pulsar.core.PulsarProducerFactory; import org.springframework.pulsar.core.PulsarTemplate; +import org.springframework.pulsar.function.PulsarFunctionAdministration; import org.springframework.pulsar.listener.AckMode; import org.springframework.pulsar.listener.PulsarContainerProperties; import org.springframework.pulsar.observation.PulsarListenerObservationConvention; @@ -245,6 +246,44 @@ class PulsarAutoConfigurationTests { } + @Nested + class FunctionAutoConfigurationTests { + + @Test + void functionSupportEnabledByDefault() { + // NOTE: hasNoNullFieldsOrProperties() ensures object providers set + contextRunner.run(context -> assertThat(context).hasNotFailed().getBean(PulsarFunctionAdministration.class) + .hasFieldOrPropertyWithValue("failFast", Boolean.TRUE) + .hasFieldOrPropertyWithValue("propagateFailures", Boolean.TRUE).hasNoNullFieldsOrProperties() + .extracting("pulsarAdministration").isSameAs(context.getBean(PulsarAdministration.class))); + } + + @Test + void functionSupportCanBeConfigured() { + contextRunner + .withPropertyValues("spring.pulsar.fu,nction.fail-fast=false", + "spring.pulsar.function.propagate-failures=false") + .run(context -> assertThat(context).hasNotFailed().getBean(PulsarFunctionAdministration.class) + .hasFieldOrPropertyWithValue("failFast", Boolean.FALSE) + .hasFieldOrPropertyWithValue("propagateFailures", Boolean.FALSE)); + } + + @Test + void functionSupportCanBeDisabled() { + contextRunner.withPropertyValues("spring.pulsar.function.enabled=false").run( + context -> assertThat(context).hasNotFailed().doesNotHaveBean(PulsarFunctionAdministration.class)); + } + + @Test + void customFunctionAdminIsRespected() { + PulsarFunctionAdministration customFunctionAdmin = mock(PulsarFunctionAdministration.class); + contextRunner.withBean(PulsarFunctionAdministration.class, () -> customFunctionAdmin) + .run(context -> assertThat(context).hasNotFailed().getBean(PulsarFunctionAdministration.class) + .isSameAs(customFunctionAdmin)); + } + + } + @Nested class ObservationAutoConfigurationTests { diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarFunctionTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarFunctionTests.java new file mode 100644 index 00000000..27c4a766 --- /dev/null +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarFunctionTests.java @@ -0,0 +1,113 @@ +/* + * 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.autoconfigure; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.catchThrowableOfType; + +import java.time.Duration; +import java.util.HashMap; +import java.util.Map; + +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.common.io.SourceConfig; +import org.assertj.core.api.InstanceOfAssertFactories; +import org.junit.jupiter.api.Test; +import org.testcontainers.containers.RabbitMQContainer; +import org.testcontainers.junit.jupiter.Container; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.WebApplicationType; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.pulsar.function.PulsarFunctionAdministration; +import org.springframework.pulsar.function.PulsarFunctionAdministration.PulsarFunctionException; +import org.springframework.pulsar.function.PulsarFunctionOperations; +import org.springframework.pulsar.function.PulsarSource; + +/** + * Integration tests for {@link PulsarFunctionAdministration}. + * + *

+ * Verifies end-to-end that a user-configured {@link PulsarSource} results in a call to + * the Pulsar broker to register the source connector. + * + * @author Chris Bono + */ +class PulsarFunctionTests implements PulsarTestContainerSupport { + + // Not currently used by anything but prepares the test for when we do verify end-end + @Container + static RabbitMQContainer rabbit = new RabbitMQContainer("rabbitmq").withExposedPorts(5672, 15672) + .withStartupTimeout(Duration.ofMinutes(1)); + + /** + * Because the docker image we use does not contain the built-in connectors we only + * verify that the configured functions are attempted to be registered w/ the broker. + * + *

+ * Later we may provide a docker image that contains the built-in connectors and + * verify the complete end-end flow. + */ + @Test + void verifyPulsarSourceIsAttemptedToBeCreatedOnBroker() { + SpringApplication app = new SpringApplication(PulsarFunctionTestConfiguration.class); + app.setWebApplicationType(WebApplicationType.NONE); + + // Again, this is a temp solution to verification of this feature + PulsarFunctionException thrown = catchThrowableOfType( + () -> app.run("--spring.pulsar.client.serviceUrl=" + PulsarTestContainerSupport.getPulsarBrokerUrl(), + "--spring.pulsar.administration.service-url=" + PulsarTestContainerSupport.getHttpServiceUrl(), + "--spring.rabbitmq.host=" + rabbit.getHost(), "--spring.rabbitmq.port=" + rabbit.getAmqpPort()), + PulsarFunctionException.class); + + Map, Exception> failures = thrown.getFailures(); + assertThat(failures).hasSize(1); + Map.Entry, Exception> failureEntry = failures.entrySet().iterator().next(); + assertThat(failureEntry.getKey()).isInstanceOf(PulsarSource.class) + .extracting("config", InstanceOfAssertFactories.type(SourceConfig.class)) + .extracting(SourceConfig::getName).isEqualTo("rabbit-test-source"); + assertThat(failureEntry.getValue()).isInstanceOf(PulsarAdminException.class) + .hasMessageContaining("Built-in source is not available"); + } + + @Configuration(proxyBeanMethods = false) + @Import(PulsarAutoConfiguration.class) + static class PulsarFunctionTestConfiguration { + + @Bean + PulsarSource rabbitSource(@Value("${spring.rabbitmq.host}") String rabbitHost, + @Value("${spring.rabbitmq.port}") int rabbitPort) { + Map configs = new HashMap<>(); + configs.put("host", rabbitHost); + configs.put("port", rabbitPort); + configs.put("virtualHost", "/"); + configs.put("username", "guest"); + configs.put("password", "guest"); + configs.put("queueName", "test_rabbit"); + configs.put("connectionName", "test-connection"); + SourceConfig sourceConfig = SourceConfig.builder().tenant("public").namespace("default") + .name("rabbit-test-source").archive("builtin://rabbitmq").topicName("incoming_rabbit") + .configs(configs).build(); + return new PulsarSource(sourceConfig, null); + } + + } + +} diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java index c33e8a55..7c51a6d3 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * 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. @@ -21,7 +21,6 @@ import static org.assertj.core.api.Assertions.assertThatIllegalArgumentException import static org.assertj.core.api.Assertions.assertThatNoException; import static org.assertj.core.api.Assertions.assertThatRuntimeException; -import java.util.Collections; import java.util.HashMap; import java.util.Map; @@ -61,10 +60,6 @@ public class PulsarPropertiesTests { private final PulsarProperties properties = new PulsarProperties(); - private void bind(String name, String value) { - bind(Collections.singletonMap(name, value)); - } - private void bind(Map map) { ConfigurationPropertySource source = new MapConfigurationPropertySource(map); new Binder(source).bind("spring.pulsar", Bindable.ofInstance(this.properties)); @@ -456,4 +451,27 @@ public class PulsarPropertiesTests { } + @Nested + class FunctionPropertiesTests { + + @Test + void functionProperties() { + Map props = new HashMap<>(); + bind(props); + + // check defaults + assertThat(properties.getFunction().getFailFast()).isTrue(); + assertThat(properties.getFunction().getPropagateFailures()).isTrue(); + + // set values and verify + props.put("spring.pulsar.function.fail-fast", "false"); + props.put("spring.pulsar.function.propagate-failures", "false"); + bind(props); + + assertThat(properties.getFunction().getFailFast()).isFalse(); + assertThat(properties.getFunction().getPropagateFailures()).isFalse(); + } + + } + } diff --git a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarTestContainerSupport.java b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarTestContainerSupport.java index d505855a..d1fd6260 100644 --- a/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarTestContainerSupport.java +++ b/spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarTestContainerSupport.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * 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. @@ -32,7 +32,7 @@ import org.testcontainers.utility.DockerImageName; public interface PulsarTestContainerSupport { PulsarContainer PULSAR_CONTAINER = new PulsarContainer( - isRunningOnMacM1() ? getMacM1PulsarImage() : getStandardPulsarImage()); + isRunningOnMacM1() ? getMacM1PulsarImage() : getStandardPulsarImage()).withFunctionsWorker(); @BeforeAll static void startContainer() { diff --git a/spring-pulsar/build.gradle b/spring-pulsar/build.gradle index ae85d6cd..52719593 100644 --- a/spring-pulsar/build.gradle +++ b/spring-pulsar/build.gradle @@ -34,6 +34,7 @@ dependencies { testImplementation 'org.hamcrest:hamcrest' testImplementation 'org.mockito:mockito-junit-jupiter' testImplementation 'org.springframework:spring-test' + testImplementation 'org.springframework.boot:spring-boot-test' testImplementation 'org.testcontainers:junit-jupiter' testImplementation 'org.testcontainers:pulsar' } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java index f5f60dcb..ba6eb1a9 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/core/PulsarAdministration.java @@ -1,5 +1,5 @@ /* - * Copyright 2022 the original author or authors. + * 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. @@ -125,7 +125,7 @@ public class PulsarAdministration createOrModifyTopicsIfNeeded(topics); } - private PulsarAdmin createAdminClient() throws PulsarClientException { + public PulsarAdmin createAdminClient() throws PulsarClientException { return this.adminBuilder.build(); } diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunction.java b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunction.java new file mode 100644 index 00000000..6634b44c --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunction.java @@ -0,0 +1,75 @@ +/* + * 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.function; + +import javax.annotation.Nullable; + +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.functions.UpdateOptions; + +/** + * Represents a user-defined Pulsar Function backed by a {@link FunctionConfig}. + * @param config the function details + * @param updateOptions the options to use during an update operation (optional) + * + * @author Chris Bono + */ +public record PulsarFunction(FunctionConfig config, + @Nullable UpdateOptions updateOptions) implements PulsarFunctionOperations { + + @Override + public String name() { + return config().getName(); + } + + @Override + public FunctionType type() { + return FunctionType.FUNCTION; + } + + @Override + public String archive() { + return config().getJar(); + } + + @Override + public FunctionConfig get(PulsarAdmin admin) throws PulsarAdminException { + return admin.functions().getFunction(config().getTenant(), config().getNamespace(), config().getName()); + } + + @Override + public void updateWithUrl(PulsarAdmin admin) throws PulsarAdminException { + admin.functions().updateFunctionWithUrl(config(), archive(), updateOptions()); + } + + @Override + public void update(PulsarAdmin admin) throws PulsarAdminException { + admin.functions().updateFunction(config(), archive(), updateOptions()); + } + + @Override + public void createWithUrl(PulsarAdmin admin) throws PulsarAdminException { + admin.functions().createFunctionWithUrl(config(), archive()); + } + + @Override + public void create(PulsarAdmin admin) throws PulsarAdminException { + admin.functions().createFunction(config(), archive()); + } +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionAdministration.java b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionAdministration.java new file mode 100644 index 00000000..30fd4265 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionAdministration.java @@ -0,0 +1,206 @@ +/* + * 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.function; + +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Optional; +import java.util.stream.Stream; + +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.common.functions.Utils; + +import org.springframework.beans.factory.ObjectProvider; +import org.springframework.beans.factory.SmartInitializingSingleton; +import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.PulsarException; +import org.springframework.pulsar.core.PulsarAdministration; + +/** + * Responsible for creating and updating any user-defined Pulsar functions, sinks, or + * sources. + * + * @author Chris Bono + */ +public class PulsarFunctionAdministration implements SmartInitializingSingleton { + + private final LogAccessor logger = new LogAccessor(this.getClass()); + + private final PulsarAdministration pulsarAdministration; + + private final ObjectProvider pulsarFunctions; + + private final ObjectProvider pulsarSinks; + + private final ObjectProvider pulsarSources; + + private final boolean failFast; + + private final boolean propagateFailures; + + /** + * Construct a {@code PulsarFunctionAdministration} instance. + * @param pulsarAdministration the pulsar admin to make the API calls with + * @param pulsarFunctions provider of functions to create/update + * @param pulsarSinks provider of sinks to create/update + * @param pulsarSources provider of sources to create/update + * @param failFast whether to stop processing when a failure occurs + * @param propagateFailures whether to throw an exception when a failure occurs + */ + public PulsarFunctionAdministration(PulsarAdministration pulsarAdministration, + ObjectProvider pulsarFunctions, ObjectProvider pulsarSinks, + ObjectProvider pulsarSources, boolean failFast, boolean propagateFailures) { + this.pulsarAdministration = pulsarAdministration; + this.pulsarFunctions = pulsarFunctions; + this.pulsarSinks = pulsarSinks; + this.pulsarSources = pulsarSources; + this.failFast = failFast; + this.propagateFailures = propagateFailures; + } + + @Override + public void afterSingletonsInstantiated() { + createOrUpdateUserDefinedFunctions(); + } + + /** + * Creates or updates any Pulsar functions registered by the application. + * + *

+ * The functions, sinks, and sources are processed serially (in that order) as + * follows: + *

    + *
  • A create or update operation is performed depending on whether or not the + * function already exists. + *
  • If the operation fails the {@link #failFast} property controls whether + * processing should stop (fail fast) or continue on w/ the next function. + *
+ * + *

+ * Once processing is complete, any failures are either logged or thrown to the caller + * (propagated) dependent on the {@link #propagateFailures} property. + * @throws PulsarFunctionException containing processing errors if the + * {@code propagateFailures} property is set to {@code true} + */ + public void createOrUpdateUserDefinedFunctions() { + try (PulsarAdmin admin = this.pulsarAdministration.createAdminClient()) { + // Concat the functions/sinks/sources into a single stream + Stream> allFunctions = Stream.concat( + Stream.concat(this.pulsarFunctions.orderedStream(), this.pulsarSinks.orderedStream()), + this.pulsarSources.orderedStream()); + + // Spin through the combined stream and process each function + Map, Exception> failures = new LinkedHashMap<>(); + for (PulsarFunctionOperations function : allFunctions.toList()) { + Optional failure = createOrUpdateFunction(function, admin); + if (failure.isPresent()) { + failures.put(function, failure.get()); + if (this.failFast) { + break; + } + } + } + + // Handle failures accordingly + if (!failures.isEmpty()) { + String msg = "Encountered " + failures.size() + " error(s) creating/updating functions: " + failures; + if (this.propagateFailures) { + throw new PulsarFunctionException(msg, failures); + } + this.logger.error(() -> msg); + } + } + catch (PulsarClientException ex) { + String msg = "Unable to create/update functions - could not create PulsarAdmin: " + ex.getMessage(); + if (this.propagateFailures) { + throw new PulsarException(msg, ex); + } + this.logger.error(ex, () -> msg); + } + } + + private Optional createOrUpdateFunction(PulsarFunctionOperations function, PulsarAdmin admin) { + try { + // Use url api for 'http|file|source|sink|function' + String archive = function.archive(); + boolean usePackageUrl = Utils.isFunctionPackageUrlSupported(archive); + if (function.functionExists(admin)) { + if (usePackageUrl) { + this.logger.info(() -> buildLogMsg(function, true, true)); + function.updateWithUrl(admin); + } + else { + this.logger.info(() -> buildLogMsg(function, true, false)); + function.update(admin); + } + } + else { + if (usePackageUrl) { + this.logger.info(() -> buildLogMsg(function, false, true)); + function.createWithUrl(admin); + } + else { + this.logger.info(() -> buildLogMsg(function, false, false)); + function.create(admin); + } + } + return Optional.empty(); + } + catch (PulsarAdminException ex) { + if (ex.getStatusCode() == 400 && "Update contains no change".equals(ex.getHttpError())) { + this.logger.debug(() -> "Update contained no change for " + functionDesc(function)); + return Optional.empty(); + } + return Optional.of(ex); + } + catch (Exception ex) { + return Optional.of(ex); + } + } + + private String buildLogMsg(PulsarFunctionOperations function, boolean isUpdate, boolean isUrlArchive) { + // '' (using (url|local) archive: + // Ex: Updating 'Uppercase' function (using url archive: sink://foo.bar) + return String.format("%s %s (using %s archive: %s)", isUpdate ? "Updating" : "Creating", functionDesc(function), + isUrlArchive ? "url" : "local", function.archive()); + } + + private String functionDesc(PulsarFunctionOperations function) { + return String.format("'%s' %s", function.name(), function.type().toString().toLowerCase()); + } + + /** + * Indicates a failure of one or more function operations. + */ + public static class PulsarFunctionException extends PulsarException { + + private final Map, Exception> failures; + + public PulsarFunctionException(String msg, Map, Exception> failures) { + super(msg); + this.failures = failures; + } + + public Map, Exception> getFailures() { + return this.failures; + } + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionOperations.java b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionOperations.java new file mode 100644 index 00000000..ca03e4ed --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionOperations.java @@ -0,0 +1,155 @@ +/* + * 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.function; + +import java.util.Optional; + +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException; +import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.io.SinkConfig; +import org.apache.pulsar.common.io.SourceConfig; + +import org.springframework.core.log.LogAccessor; + +/** + * Provides operations for a particular function type. + * + *

Function Types

+ * + *

+ * The term 'function' is meant to cover Pulsar IO connectors (sources and sinks) as well + * as Pulsar Functions (user defined functions). + * + *

Motivation

+ * + *

+ * The underlying Pulsar function model treats source, sink, and function completely + * separately, including its config objects and API calls. This operations abstraction + * allows them to be treated as related during processing. + * + * @param the function config (one of {@link SourceConfig}, {@link SinkConfig}, + * {@link FunctionConfig}) + * @author Chris Bono + */ +public interface PulsarFunctionOperations { + + /** + * Logger used in default methods. + */ + LogAccessor logger = new LogAccessor(PulsarFunctionOperations.class); + + /** + * Gets the name of the function. + * @return the name of the function + */ + String name(); + + /** + * Gets the type of function the operations handles. + * @return the type of the function + */ + FunctionType type(); + + /** + * Gets the url or path to the archive that represents the function. + * @return the url or path to the archive that represents function + */ + String archive(); + + /** + * Gets the configuration details for an existing function. + * @param admin the admin client + * @return the current config of the existing function + * @throws NotFoundException if function does not exist + * @throws PulsarAdminException if anything else goes wrong + */ + T get(PulsarAdmin admin) throws PulsarAdminException; + + /** + * Updates the function using the url-based update api. + * @param admin the admin client + * @throws PulsarAdminException if anything else goes wrong + */ + void updateWithUrl(PulsarAdmin admin) throws PulsarAdminException; + + /** + * Updates the function using the file-based update api. + * @param admin the admin client + * @throws PulsarAdminException if anything else goes wrong + */ + void update(PulsarAdmin admin) throws PulsarAdminException; + + /** + * Creates the function using the url-based create api. + * @param admin the admin client + * @throws PulsarAdminException if anything else goes wrong + */ + void createWithUrl(PulsarAdmin admin) throws PulsarAdminException; + + /** + * Creates the function using the file-based create api. + * @param admin the admin client + * @throws PulsarAdminException if anything else goes wrong + */ + void create(PulsarAdmin admin) throws PulsarAdminException; + + /** + * Determines if a function already exists. + * @param admin the admin client + * @return {@code true} if function already exists + * @throws PulsarAdminException if anything else goes wrong + */ + default boolean functionExists(PulsarAdmin admin) throws PulsarAdminException { + return getIfExists(admin).isPresent(); + } + + /** + * Gets the configuration details for a function if it exists. + * @param admin the admin client + * @return the optional current config of the function or empty if the function does + * not exist + * @throws PulsarAdminException if anything else goes wrong + */ + default Optional getIfExists(PulsarAdmin admin) throws PulsarAdminException { + try { + return Optional.of(get(admin)); + } + catch (NotFoundException ex) { + logger.trace(ex, () -> "Function not found: " + name()); + return Optional.empty(); + } + } + + /** + * The type of function the operations handle. + */ + enum FunctionType { + + /** A user-defined Pulsar function. */ + FUNCTION, + + /** A Pulsar sink connector. */ + SINK, + + /** A Pulsar source connector. */ + SOURCE + + } + +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSink.java b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSink.java new file mode 100644 index 00000000..ac0b6375 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSink.java @@ -0,0 +1,75 @@ +/* + * 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.function; + +import javax.annotation.Nullable; + +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.common.functions.UpdateOptions; +import org.apache.pulsar.common.io.SinkConfig; + +/** + * Represents a Pulsar Sink backed by a {@link SinkConfig}. + * @param config the sink details + * @param updateOptions the options to use during an update operation (optional) + * + * @author Chris Bono + */ +public record PulsarSink(SinkConfig config, + @Nullable UpdateOptions updateOptions) implements PulsarFunctionOperations { + + @Override + public String name() { + return config().getName(); + } + + @Override + public FunctionType type() { + return FunctionType.SINK; + } + + @Override + public String archive() { + return config().getArchive(); + } + + @Override + public SinkConfig get(PulsarAdmin admin) throws PulsarAdminException { + return admin.sinks().getSink(config().getTenant(), config().getNamespace(), config().getName()); + } + + @Override + public void updateWithUrl(PulsarAdmin admin) throws PulsarAdminException { + admin.sinks().updateSinkWithUrl(config(), archive(), updateOptions()); + } + + @Override + public void update(PulsarAdmin admin) throws PulsarAdminException { + admin.sinks().updateSink(config(), archive(), updateOptions()); + } + + @Override + public void createWithUrl(PulsarAdmin admin) throws PulsarAdminException { + admin.sinks().createSinkWithUrl(config(), archive()); + } + + @Override + public void create(PulsarAdmin admin) throws PulsarAdminException { + admin.sinks().createSink(config(), archive()); + } +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSource.java b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSource.java new file mode 100644 index 00000000..9b436076 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSource.java @@ -0,0 +1,75 @@ +/* + * 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.function; + +import javax.annotation.Nullable; + +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.common.functions.UpdateOptions; +import org.apache.pulsar.common.io.SourceConfig; + +/** + * Represents a Pulsar Source backed by a {@link SourceConfig}. + * @param config the source details + * @param updateOptions the options to use during an update operation (optional) + * + * @author Chris Bono + */ +public record PulsarSource(SourceConfig config, + @Nullable UpdateOptions updateOptions) implements PulsarFunctionOperations { + + @Override + public String name() { + return config().getName(); + } + + @Override + public FunctionType type() { + return FunctionType.SOURCE; + } + + @Override + public String archive() { + return config().getArchive(); + } + + @Override + public SourceConfig get(PulsarAdmin admin) throws PulsarAdminException { + return admin.sources().getSource(config().getTenant(), config().getNamespace(), config().getName()); + } + + @Override + public void updateWithUrl(PulsarAdmin admin) throws PulsarAdminException { + admin.sources().updateSourceWithUrl(config(), archive(), updateOptions()); + } + + @Override + public void update(PulsarAdmin admin) throws PulsarAdminException { + admin.sources().updateSource(config(), archive(), updateOptions()); + } + + @Override + public void createWithUrl(PulsarAdmin admin) throws PulsarAdminException { + admin.sources().createSourceWithUrl(config(), archive()); + } + + @Override + public void create(PulsarAdmin admin) throws PulsarAdminException { + admin.sources().createSource(config(), archive()); + } +} diff --git a/spring-pulsar/src/main/java/org/springframework/pulsar/function/package-info.java b/spring-pulsar/src/main/java/org/springframework/pulsar/function/package-info.java new file mode 100644 index 00000000..68f7a452 --- /dev/null +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/package-info.java @@ -0,0 +1,12 @@ +/** + * Package containing support for + * Pulsar + * Functions and Pulsar + * IO Connectors. + */ +@NonNullApi +@NonNullFields +package org.springframework.pulsar.function; + +import org.springframework.lang.NonNullApi; +import org.springframework.lang.NonNullFields; diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionAdministrationTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionAdministrationTests.java new file mode 100644 index 00000000..f6630267 --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionAdministrationTests.java @@ -0,0 +1,376 @@ +/* + * 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.function; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.assertj.core.api.Assertions.catchThrowableOfType; +import static org.assertj.core.api.Assertions.entry; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.verifyNoMoreInteractions; +import static org.mockito.Mockito.when; + +import org.apache.pulsar.client.admin.PulsarAdmin; +import org.apache.pulsar.client.admin.PulsarAdminException; +import org.apache.pulsar.client.admin.PulsarAdminException.NotFoundException; +import org.apache.pulsar.client.api.PulsarClientException; +import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.functions.UpdateOptions; +import org.apache.pulsar.common.functions.UpdateOptionsImpl; +import org.apache.pulsar.common.io.SinkConfig; +import org.apache.pulsar.common.io.SourceConfig; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mockito; + +import org.springframework.beans.factory.support.StaticListableBeanFactory; +import org.springframework.boot.test.system.CapturedOutput; +import org.springframework.boot.test.system.OutputCaptureExtension; +import org.springframework.pulsar.PulsarException; +import org.springframework.pulsar.core.PulsarAdministration; +import org.springframework.pulsar.function.PulsarFunctionAdministration.PulsarFunctionException; + +/** + * Tests for {@link PulsarFunctionAdministration}. + * + * @author Chris Bono + */ +class PulsarFunctionAdministrationTests { + + private PulsarAdmin pulsarAdmin = mock(PulsarAdmin.class, Mockito.RETURNS_DEEP_STUBS); + + private PulsarAdministration springPulsarAdmin = mock(PulsarAdministration.class); + + private PulsarFunctionAdministration functionAdmin; + + private StaticListableBeanFactory beanFactory; + + @BeforeEach + void setupAdminsAndBeanFactory() throws PulsarClientException { + when(springPulsarAdmin.createAdminClient()).thenReturn(pulsarAdmin); + beanFactory = new StaticListableBeanFactory(); + functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, + beanFactory.getBeanProvider(PulsarFunction.class), beanFactory.getBeanProvider(PulsarSink.class), + beanFactory.getBeanProvider(PulsarSource.class), true, true); + } + + @Test + void noFunctionsSinksOrSourcesProvided() throws PulsarClientException { + functionAdmin.createOrUpdateUserDefinedFunctions(); + verify(springPulsarAdmin).createAdminClient(); + verifyNoMoreInteractions(springPulsarAdmin); + verify(pulsarAdmin).close(); + verifyNoMoreInteractions(pulsarAdmin); + } + + @Nested + class ManagePulsarFunctions { + + private FunctionConfig functionConfig = FunctionConfig.builder().tenant("tenant1").namespace("namespace1") + .name("function1").jar("function1.jar").build(); + + @Test + void createFunction() throws PulsarAdminException { + when(pulsarAdmin.functions().getFunction("tenant1", "namespace1", "function1")) + .thenThrow(new NotFoundException(null, "400", 400)); + beanFactory.addBean("myFunction", new PulsarFunction(functionConfig, null)); + + functionAdmin.createOrUpdateUserDefinedFunctions(); + + verify(pulsarAdmin.functions()).getFunction(functionConfig.getTenant(), functionConfig.getNamespace(), + functionConfig.getName()); + verify(pulsarAdmin.functions()).createFunction(functionConfig, functionConfig.getJar()); + } + + @Test + void updateFunction() throws PulsarAdminException { + when(pulsarAdmin.functions().getFunction("tenant1", "namespace1", "function1")).thenReturn(functionConfig); + FunctionConfig functionConfigNew = functionConfig.toBuilder().jar("function1-v2.jar").build(); + beanFactory.addBean("myFunction", new PulsarFunction(functionConfigNew, null)); + + functionAdmin.createOrUpdateUserDefinedFunctions(); + + verify(pulsarAdmin.functions()).getFunction(functionConfigNew.getTenant(), functionConfigNew.getNamespace(), + functionConfigNew.getName()); + verify(pulsarAdmin.functions()).updateFunction(functionConfigNew, functionConfigNew.getJar(), null); + } + + } + + @Nested + class ManagePulsarSinks { + + private SinkConfig sinkConfig = SinkConfig.builder().tenant("tenant1").namespace("namespace1").name("sink1") + .archive("sink1.jar").build(); + + @Test + void createSink() throws PulsarAdminException { + when(pulsarAdmin.sinks().getSink("tenant1", "namespace1", "sink1")) + .thenThrow(new NotFoundException(null, "400", 400)); + + beanFactory.addBean("mySink", new PulsarSink(sinkConfig, null)); + + functionAdmin.createOrUpdateUserDefinedFunctions(); + + verify(pulsarAdmin.sinks()).getSink(sinkConfig.getTenant(), sinkConfig.getNamespace(), + sinkConfig.getName()); + verify(pulsarAdmin.sinks()).createSink(sinkConfig, sinkConfig.getArchive()); + } + + @Test + void updateSink() throws PulsarAdminException { + when(pulsarAdmin.sinks().getSink("tenant1", "namespace1", "sink1")).thenReturn(sinkConfig); + SinkConfig sinkConfigNew = sinkConfig.toBuilder().archive("sink1-v2.jar").build(); + UpdateOptions updateOptions = new UpdateOptionsImpl(); + beanFactory.addBean("mySink", new PulsarSink(sinkConfigNew, updateOptions)); + + functionAdmin.createOrUpdateUserDefinedFunctions(); + + verify(pulsarAdmin.sinks()).getSink(sinkConfigNew.getTenant(), sinkConfigNew.getNamespace(), + sinkConfigNew.getName()); + verify(pulsarAdmin.sinks()).updateSink(sinkConfigNew, sinkConfigNew.getArchive(), updateOptions); + } + + } + + @Nested + class ManagePulsarSources { + + private SourceConfig sourceConfig = SourceConfig.builder().tenant("tenant1").namespace("namespace1") + .name("source1").archive("source1.jar").build(); + + @Test + void createSource() throws PulsarAdminException { + when(pulsarAdmin.sources().getSource("tenant1", "namespace1", "source1")) + .thenThrow(new NotFoundException(null, "400", 400)); + beanFactory.addBean("mySource", new PulsarSource(sourceConfig, null)); + + functionAdmin.createOrUpdateUserDefinedFunctions(); + + verify(pulsarAdmin.sources()).getSource(sourceConfig.getTenant(), sourceConfig.getNamespace(), + sourceConfig.getName()); + verify(pulsarAdmin.sources()).createSource(sourceConfig, sourceConfig.getArchive()); + } + + @Test + void updateSource() throws PulsarAdminException { + when(pulsarAdmin.sources().getSource("tenant1", "namespace1", "source1")).thenReturn(sourceConfig); + SourceConfig sourceConfigNew = sourceConfig.toBuilder().archive("source1-v2.jar").build(); + UpdateOptions updateOptions = new UpdateOptionsImpl(); + beanFactory.addBean("mySource", new PulsarSource(sourceConfigNew, updateOptions)); + + functionAdmin.createOrUpdateUserDefinedFunctions(); + + verify(pulsarAdmin.sources()).getSource(sourceConfigNew.getTenant(), sourceConfigNew.getNamespace(), + sourceConfigNew.getName()); + verify(pulsarAdmin.sources()).updateSource(sourceConfigNew, sourceConfigNew.getArchive(), updateOptions); + } + + } + + @Nested + class ProcessHandling { + + private PulsarFunction function1; + + private PulsarSink sink1; + + private PulsarSource source1; + + @BeforeEach + void setupFunctionsSinksAndSources() throws PulsarAdminException { + + function1 = mock(PulsarFunction.class); + when(function1.functionExists(pulsarAdmin)).thenReturn(false); + + sink1 = mock(PulsarSink.class); + when(sink1.functionExists(pulsarAdmin)).thenReturn(false); + + source1 = mock(PulsarSource.class); + when(source1.functionExists(pulsarAdmin)).thenReturn(false); + + beanFactory.addBean("function1", function1); + beanFactory.addBean("sink1", sink1); + beanFactory.addBean("source1", source1); + } + + @Test + void allFunctionsProcessedSuccessfully() throws PulsarAdminException { + functionAdmin.createOrUpdateUserDefinedFunctions(); + verify(function1).create(pulsarAdmin); + verify(sink1).create(pulsarAdmin); + verify(source1).create(pulsarAdmin); + } + + @Test + void firstProcessedFunctionFailsFast() throws PulsarAdminException { + PulsarAdminException ex = new PulsarAdminException("BOOM"); + when(function1.functionExists(pulsarAdmin)).thenThrow(ex); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.createOrUpdateUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(function1, ex)); + verify(function1, never()).create(pulsarAdmin); + verify(function1, never()).update(pulsarAdmin); + verifyNoInteractions(sink1, source1); + } + + @Test + void middleProcessedFunctionFailsFast() throws PulsarAdminException { + PulsarAdminException ex = new PulsarAdminException("BOOM"); + when(sink1.functionExists(pulsarAdmin)).thenThrow(ex); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.createOrUpdateUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(sink1, ex)); + verify(function1).create(pulsarAdmin); + verify(sink1, never()).create(pulsarAdmin); + verify(sink1, never()).update(pulsarAdmin); + verifyNoInteractions(source1); + } + + @Test + void lastProcessedFunctionFailsFast() throws PulsarAdminException { + PulsarAdminException ex = new PulsarAdminException("BOOM"); + when(source1.functionExists(pulsarAdmin)).thenThrow(ex); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.createOrUpdateUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(source1, ex)); + verify(function1).create(pulsarAdmin); + verify(sink1).create(pulsarAdmin); + verify(source1, never()).create(pulsarAdmin); + verify(source1, never()).update(pulsarAdmin); + } + + @Test + void firstProcessedFunctionFailsSlow() throws PulsarAdminException { + functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, + beanFactory.getBeanProvider(PulsarFunction.class), beanFactory.getBeanProvider(PulsarSink.class), + beanFactory.getBeanProvider(PulsarSource.class), false, true); + PulsarAdminException ex = new PulsarAdminException("BOOM"); + when(function1.functionExists(pulsarAdmin)).thenThrow(ex); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.createOrUpdateUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(function1, ex)); + verify(function1, never()).create(pulsarAdmin); + verify(function1, never()).update(pulsarAdmin); + verify(sink1).create(pulsarAdmin); + verify(source1).create(pulsarAdmin); + } + + @Test + void middleProcessedFunctionFailsSlow() throws PulsarAdminException { + functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, + beanFactory.getBeanProvider(PulsarFunction.class), beanFactory.getBeanProvider(PulsarSink.class), + beanFactory.getBeanProvider(PulsarSource.class), false, true); + PulsarAdminException ex = new PulsarAdminException("BOOM"); + when(sink1.functionExists(pulsarAdmin)).thenThrow(ex); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.createOrUpdateUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(sink1, ex)); + verify(function1).create(pulsarAdmin); + verify(sink1, never()).create(pulsarAdmin); + verify(sink1, never()).update(pulsarAdmin); + verify(source1).create(pulsarAdmin); + } + + @Test + void lastProcessedFunctionFailsSlow() throws PulsarAdminException { + functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, + beanFactory.getBeanProvider(PulsarFunction.class), beanFactory.getBeanProvider(PulsarSink.class), + beanFactory.getBeanProvider(PulsarSource.class), false, true); + PulsarAdminException ex = new PulsarAdminException("BOOM"); + when(source1.functionExists(pulsarAdmin)).thenThrow(ex); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.createOrUpdateUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(source1, ex)); + verify(function1).create(pulsarAdmin); + verify(sink1).create(pulsarAdmin); + verify(source1, never()).create(pulsarAdmin); + verify(source1, never()).update(pulsarAdmin); + } + + @Test + void allProcessedFunctionsFailSlow() throws PulsarAdminException { + functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, + beanFactory.getBeanProvider(PulsarFunction.class), beanFactory.getBeanProvider(PulsarSink.class), + beanFactory.getBeanProvider(PulsarSource.class), false, true); + PulsarAdminException ex1 = new PulsarAdminException("BOOM1"); + PulsarAdminException ex2 = new PulsarAdminException("BOOM2"); + PulsarAdminException ex3 = new PulsarAdminException("BOOM3"); + when(function1.functionExists(pulsarAdmin)).thenThrow(ex1); + when(sink1.functionExists(pulsarAdmin)).thenThrow(ex2); + when(source1.functionExists(pulsarAdmin)).thenThrow(ex3); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.createOrUpdateUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(function1, ex1), entry(sink1, ex2), + entry(source1, ex3)); + verify(function1, never()).create(pulsarAdmin); + verify(function1, never()).update(pulsarAdmin); + verify(sink1, never()).create(pulsarAdmin); + verify(sink1, never()).update(pulsarAdmin); + verify(source1, never()).create(pulsarAdmin); + verify(source1, never()).update(pulsarAdmin); + } + + @Test + void createAdminClientFails() throws PulsarClientException { + when(springPulsarAdmin.createAdminClient()).thenThrow(new PulsarClientException("NOPE")); + assertThatThrownBy(() -> functionAdmin.createOrUpdateUserDefinedFunctions()) + .isInstanceOf(PulsarException.class) + .hasMessageContaining("Unable to create/update functions - could not create PulsarAdmin: NOPE"); + } + + } + + @Nested + @ExtendWith(OutputCaptureExtension.class) + class ProcessHandlingPropagationDisabled { + + @BeforeEach + void setupFunctionAdminWithPropagationDisabled() { + functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, + beanFactory.getBeanProvider(PulsarFunction.class), beanFactory.getBeanProvider(PulsarSink.class), + beanFactory.getBeanProvider(PulsarSource.class), true, false); + } + + @Test + void createAdminClientFails(CapturedOutput output) throws PulsarClientException { + when(springPulsarAdmin.createAdminClient()).thenThrow(new PulsarClientException("NOPE")); + functionAdmin.createOrUpdateUserDefinedFunctions(); + assertThat(output).contains("Unable to create/update functions - could not create PulsarAdmin: NOPE"); + } + + @Test + void processedFunctionFails(CapturedOutput output) throws PulsarAdminException { + PulsarFunction function1 = mock(PulsarFunction.class); + beanFactory.addBean("function1", function1); + PulsarAdminException ex = new PulsarAdminException("BOOM"); + when(function1.functionExists(pulsarAdmin)).thenThrow(ex); + + functionAdmin.createOrUpdateUserDefinedFunctions(); + + assertThat(output).contains("Encountered 1 error(s) creating/updating functions:", + "PulsarAdminException: BOOM"); + } + + } + +} diff --git a/spring-pulsar/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker b/spring-pulsar/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker new file mode 100644 index 00000000..1f0955d4 --- /dev/null +++ b/spring-pulsar/src/test/resources/mockito-extensions/org.mockito.plugins.MockMaker @@ -0,0 +1 @@ +mock-maker-inline