From 66beedef864e5e4e4f9139e6c4794abac3db4fd0 Mon Sep 17 00:00:00 2001 From: Chris Bono Date: Fri, 13 Jan 2023 16:04:52 -0600 Subject: [PATCH] Add stop policy for Pulsar Functions (#271) --- .../src/main/asciidoc/pulsar-function.adoc | 6 +- .../sample/signup/config/AppConfig.java | 7 +- .../PulsarAutoConfiguration.java | 3 +- .../autoconfigure/PulsarProperties.java | 17 +- .../PulsarAutoConfigurationTests.java | 11 +- .../autoconfigure/PulsarFunctionTests.java | 13 +- .../autoconfigure/PulsarPropertiesTests.java | 3 + .../pulsar/function/PulsarFunction.java | 29 +- .../PulsarFunctionAdministration.java | 162 ++++- .../function/PulsarFunctionOperations.java | 107 ++- .../pulsar/function/PulsarSink.java | 29 +- .../pulsar/function/PulsarSource.java | 29 +- .../PulsarFunctionAdministrationTests.java | 647 +++++++++++------- .../pulsar/function/PulsarFunctionsTests.java | 382 +++++++++++ 14 files changed, 1137 insertions(+), 308 deletions(-) create mode 100644 spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionsTests.java diff --git a/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc b/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc index cc88f6ac..2705444b 100644 --- a/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc +++ b/spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc @@ -27,7 +27,7 @@ By default, the application tries to connect to a local Pulsar instance at `http 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. +On application startup, 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. @@ -35,6 +35,8 @@ NOTE: The `PulsarFunction`, `PulsarSink`, and `PulsarSource` beans are simple wr 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. +On application shutdown, all functions that were processed during application startup have their stop policy enforced and are either left alone, stopped, or deleted from the Pulsar server. + == Limitations === No Magic Pulsar Functions @@ -148,3 +150,5 @@ PulsarSource rabbitSourceWithBootProps(RabbitProperties props) { } ---- ==== + +TIP: For a more elaborate example see the link:{github}/blob/main/spring-pulsar-sample-apps/sample-pulsar-functions/README.adoc[Sample Stream Pipeline with Pulsar Functions] sample app diff --git a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/config/AppConfig.java b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/config/AppConfig.java index 210c03a4..b572d376 100644 --- a/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/config/AppConfig.java +++ b/spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/config/AppConfig.java @@ -29,6 +29,7 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.pulsar.function.PulsarFunction; +import org.springframework.pulsar.function.PulsarFunctionOperations.FunctionStopPolicy; import org.springframework.pulsar.function.PulsarSink; import org.springframework.pulsar.function.PulsarSource; import org.springframework.pulsar.sample.signup.model.SignupGenerator; @@ -59,7 +60,7 @@ class AppConfig { SourceConfig sourceConfig = SourceConfig.builder().tenant("public").namespace("default") .name("UserSignupRabbitSource").archive("builtin://rabbitmq").topicName("user-signup").configs(configs) .build(); - return new PulsarSource(sourceConfig, null); + return new PulsarSource(sourceConfig, FunctionStopPolicy.DELETE, null); } @Bean @@ -74,7 +75,7 @@ class AppConfig { FunctionConfig functionConfig = FunctionConfig.builder().tenant("public").namespace("default") .name("UserSignupFunction").className("org.springframework.pulsar.sample.signup.SignupFunction") .jar(absPathToFunctionJar).inputs(List.of("user-signup")).build(); - return new PulsarFunction(functionConfig, null); + return new PulsarFunction(functionConfig, FunctionStopPolicy.DELETE, null); } @Bean @@ -88,7 +89,7 @@ class AppConfig { SinkConfig sinkConfig = SinkConfig.builder().tenant("public").namespace("default") .name("CustomerOnboardCassandraSink").archive("builtin://cassandra").inputs(List.of("customer-onboard")) .configs(configs).build(); - return new PulsarSink(sinkConfig, null); + return new PulsarSink(sinkConfig, FunctionStopPolicy.DELETE, null); } } 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 fe43cb59..08f6e34c 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 @@ -124,7 +124,8 @@ public class PulsarAutoConfiguration { ObjectProvider pulsarFunctions, ObjectProvider pulsarSinks, ObjectProvider pulsarSources) { return new PulsarFunctionAdministration(pulsarAdministration, pulsarFunctions, pulsarSinks, pulsarSources, - this.properties.getFunction().getFailFast(), this.properties.getFunction().getPropagateFailures()); + this.properties.getFunction().getFailFast(), this.properties.getFunction().getPropagateFailures(), + this.properties.getFunction().getPropagateStopFailures()); } } 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 8e3493a4..d673c300 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 @@ -1669,10 +1669,17 @@ public class PulsarProperties { private Boolean failFast = Boolean.TRUE; /** - * Whether to throw an exception if any failure is encountered during processing. + * Whether to throw an exception if any failure is encountered during server + * startup while creating/updating functions. */ private Boolean propagateFailures = Boolean.TRUE; + /** + * Whether to throw an exception if any failure is encountered during server + * shutdown while enforcing stop policy on functions. + */ + private Boolean propagateStopFailures = Boolean.FALSE; + public Boolean getFailFast() { return this.failFast; } @@ -1689,6 +1696,14 @@ public class PulsarProperties { this.propagateFailures = propagateFailures; } + public Boolean getPropagateStopFailures() { + return this.propagateStopFailures; + } + + public void setPropagateStopFailures(Boolean propagateStopFailures) { + this.propagateStopFailures = propagateStopFailures; + } + } public static class Listener { 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 ff38b510..bb0d90a1 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 @@ -254,18 +254,21 @@ class PulsarAutoConfigurationTests { // NOTE: hasNoNullFieldsOrProperties() ensures object providers set contextRunner.run(context -> assertThat(context).hasNotFailed().getBean(PulsarFunctionAdministration.class) .hasFieldOrPropertyWithValue("failFast", Boolean.TRUE) - .hasFieldOrPropertyWithValue("propagateFailures", Boolean.TRUE).hasNoNullFieldsOrProperties() + .hasFieldOrPropertyWithValue("propagateFailures", Boolean.TRUE) + .hasFieldOrPropertyWithValue("propagateStopFailures", Boolean.FALSE).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") + .withPropertyValues("spring.pulsar.function.fail-fast=false", + "spring.pulsar.function.propagate-failures=false", + "spring.pulsar.function.propagate-stop-failures=true") .run(context -> assertThat(context).hasNotFailed().getBean(PulsarFunctionAdministration.class) .hasFieldOrPropertyWithValue("failFast", Boolean.FALSE) - .hasFieldOrPropertyWithValue("propagateFailures", Boolean.FALSE)); + .hasFieldOrPropertyWithValue("propagateFailures", Boolean.FALSE) + .hasFieldOrPropertyWithValue("propagateStopFailures", Boolean.TRUE)); } @Test 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 index 27c4a766..b2b39d5f 100644 --- 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 @@ -33,12 +33,14 @@ 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.ApplicationContextException; 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.PulsarFunctionOperations.FunctionStopPolicy; import org.springframework.pulsar.function.PulsarSource; /** @@ -71,13 +73,14 @@ class PulsarFunctionTests implements PulsarTestContainerSupport { app.setWebApplicationType(WebApplicationType.NONE); // Again, this is a temp solution to verification of this feature - PulsarFunctionException thrown = catchThrowableOfType( + ApplicationContextException 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(); + ApplicationContextException.class); + assertThat(thrown).hasCauseInstanceOf(PulsarFunctionException.class); + PulsarFunctionException cause = (PulsarFunctionException) thrown.getCause(); + Map, Exception> failures = cause.getFailures(); assertThat(failures).hasSize(1); Map.Entry, Exception> failureEntry = failures.entrySet().iterator().next(); assertThat(failureEntry.getKey()).isInstanceOf(PulsarSource.class) @@ -105,7 +108,7 @@ class PulsarFunctionTests implements PulsarTestContainerSupport { 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); + return new PulsarSource(sourceConfig, FunctionStopPolicy.NONE, 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 7c51a6d3..3db98c5d 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 @@ -462,14 +462,17 @@ public class PulsarPropertiesTests { // check defaults assertThat(properties.getFunction().getFailFast()).isTrue(); assertThat(properties.getFunction().getPropagateFailures()).isTrue(); + assertThat(properties.getFunction().getPropagateStopFailures()).isFalse(); // set values and verify props.put("spring.pulsar.function.fail-fast", "false"); props.put("spring.pulsar.function.propagate-failures", "false"); + props.put("spring.pulsar.function.propagate-stop-failures", "true"); bind(props); assertThat(properties.getFunction().getFailFast()).isFalse(); assertThat(properties.getFunction().getPropagateFailures()).isFalse(); + assertThat(properties.getFunction().getPropagateStopFailures()).isTrue(); } } 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 index 6634b44c..e833d5a7 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunction.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunction.java @@ -23,16 +23,23 @@ import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.functions.FunctionConfig; import org.apache.pulsar.common.functions.UpdateOptions; +import org.springframework.pulsar.PulsarException; + /** * Represents a user-defined Pulsar Function backed by a {@link FunctionConfig}. * @param config the function details + * @param stopPolicy the action to take on the function when the server is stopped * @param updateOptions the options to use during an update operation (optional) * * @author Chris Bono */ -public record PulsarFunction(FunctionConfig config, +public record PulsarFunction(FunctionConfig config, FunctionStopPolicy stopPolicy, @Nullable UpdateOptions updateOptions) implements PulsarFunctionOperations { + public PulsarFunction(FunctionConfig config, @Nullable UpdateOptions updateOptions) { + this(config, FunctionStopPolicy.DELETE, updateOptions); + } + @Override public String name() { return config().getName(); @@ -72,4 +79,24 @@ public record PulsarFunction(FunctionConfig config, public void create(PulsarAdmin admin) throws PulsarAdminException { admin.functions().createFunction(config(), archive()); } + + @Override + public void stop(PulsarAdmin admin) { + try { + admin.functions().stopFunction(config().getTenant(), config().getNamespace(), config().getName()); + } + catch (PulsarAdminException e) { + throw new PulsarException(e.getMessage(), e); + } + } + + @Override + public void delete(PulsarAdmin admin) { + try { + admin.functions().deleteFunction(config().getTenant(), config().getNamespace(), config().getName()); + } + catch (PulsarAdminException e) { + throw new PulsarException(e.getMessage(), e); + } + } } 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 index 30fd4265..a92fd600 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionAdministration.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionAdministration.java @@ -16,7 +16,10 @@ package org.springframework.pulsar.function; +import java.util.ArrayList; +import java.util.Collections; import java.util.LinkedHashMap; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.stream.Stream; @@ -27,7 +30,7 @@ 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.context.SmartLifecycle; import org.springframework.core.log.LogAccessor; import org.springframework.pulsar.PulsarException; import org.springframework.pulsar.core.PulsarAdministration; @@ -38,7 +41,7 @@ import org.springframework.pulsar.core.PulsarAdministration; * * @author Chris Bono */ -public class PulsarFunctionAdministration implements SmartInitializingSingleton { +public class PulsarFunctionAdministration implements SmartLifecycle { private final LogAccessor logger = new LogAccessor(this.getClass()); @@ -50,10 +53,16 @@ public class PulsarFunctionAdministration implements SmartInitializingSingleton private final ObjectProvider pulsarSources; + private final List> processedFunctions; + private final boolean failFast; private final boolean propagateFailures; + private final boolean propagateStopFailures; + + private volatile boolean running; + /** * Construct a {@code PulsarFunctionAdministration} instance. * @param pulsarAdministration the pulsar admin to make the API calls with @@ -61,26 +70,57 @@ public class PulsarFunctionAdministration implements SmartInitializingSingleton * @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 + * @param propagateFailures whether to throw an exception when a failure occurs during + * server startup while creating/updating functions + * @param propagateStopFailures whether to throw an exception when a failure occurs + * during server shutdown while enforcing stop policy on functions */ public PulsarFunctionAdministration(PulsarAdministration pulsarAdministration, ObjectProvider pulsarFunctions, ObjectProvider pulsarSinks, - ObjectProvider pulsarSources, boolean failFast, boolean propagateFailures) { + ObjectProvider pulsarSources, boolean failFast, boolean propagateFailures, + boolean propagateStopFailures) { this.pulsarAdministration = pulsarAdministration; this.pulsarFunctions = pulsarFunctions; this.pulsarSinks = pulsarSinks; this.pulsarSources = pulsarSources; this.failFast = failFast; this.propagateFailures = propagateFailures; + this.propagateStopFailures = propagateStopFailures; + this.processedFunctions = new ArrayList<>(); } @Override - public void afterSingletonsInstantiated() { - createOrUpdateUserDefinedFunctions(); + public synchronized void start() { + if (!this.running) { + this.logger.debug(() -> "Processing Pulsar Functions"); + long start = System.currentTimeMillis(); + this.createOrUpdateUserDefinedFunctions(); + this.running = true; + long duration = System.currentTimeMillis() - start; + this.logger.debug(() -> "Processed Pulsar Functions in " + duration + " ms"); + } + } + + @Override + public synchronized void stop() { + if (this.running) { + this.logger.debug(() -> "Enforcing stop policy on Pulsar Functions"); + this.running = false; + long start = System.currentTimeMillis(); + this.enforceStopPolicyOnUserDefinedFunctions(); + long duration = System.currentTimeMillis() - start; + this.logger.debug(() -> "Enforced stop policy on Pulsar Functions in " + duration + " ms"); + } + } + + @Override + public boolean isRunning() { + return this.running; } /** - * Creates or updates any Pulsar functions registered by the application. + * Called during server startup, creates or updates any Pulsar functions registered by + * the application. * *

* The functions, sinks, and sources are processed serially (in that order) as @@ -99,17 +139,24 @@ public class PulsarFunctionAdministration implements SmartInitializingSingleton * {@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()); + // Concat the functions/sinks/sources into a single stream + Stream> allFunctions = Stream.concat( + Stream.concat(this.pulsarFunctions.orderedStream(), this.pulsarSinks.orderedStream()), + this.pulsarSources.orderedStream()); + List> functionsToProcess = allFunctions.toList(); + if (functionsToProcess.isEmpty()) { + this.logger.debug("No user defined functions to process."); + return; + } - // Spin through the combined stream and process each function + try (PulsarAdmin admin = this.pulsarAdministration.createAdminClient()) { Map, Exception> failures = new LinkedHashMap<>(); - for (PulsarFunctionOperations function : allFunctions.toList()) { + for (PulsarFunctionOperations function : functionsToProcess) { Optional failure = createOrUpdateFunction(function, admin); - if (failure.isPresent()) { + if (failure.isEmpty()) { + this.processedFunctions.add(function); + } + else { failures.put(function, failure.get()); if (this.failFast) { break; @@ -181,6 +228,91 @@ public class PulsarFunctionAdministration implements SmartInitializingSingleton isUrlArchive ? "url" : "local", function.archive()); } + // VisibleForTesting + List> getProcessedFunctions() { + return this.processedFunctions; + } + + /** + * Called during server shutdown, enforces the stop policy on all Pulsar functions + * that were successfully processed during server startup. + * + *

+ * The functions, sinks, and sources are processed in reverse startup order as + * follows: + *

    + *
  • The stop policy of each function is used to determine if the function should be + * stopped, removed, or left alone. + *
+ * + *

+ * Once processing is complete, any failures are either logged or thrown to the caller + * (propagated) dependent on the {@link #propagateStopFailures} property. + * @throws PulsarFunctionException containing processing errors if the + * {@code propagateStopFailures} property is set to {@code true} + */ + public void enforceStopPolicyOnUserDefinedFunctions() { + if (this.processedFunctions.isEmpty()) { + this.logger.debug("No processed functions to enforce stop policy on"); + return; + } + + try (PulsarAdmin admin = this.pulsarAdministration.createAdminClient()) { + Map, Exception> failures = new LinkedHashMap<>(); + // Spin through the processed functions in reverse startup order + Collections.reverse(this.processedFunctions); + for (PulsarFunctionOperations function : this.processedFunctions) { + Optional failure = enforceStopPolicyOnFunction(function, admin); + failure.ifPresent(e -> failures.put(function, e)); + } + + // Handle failures accordingly + if (!failures.isEmpty()) { + String msg = "Encountered " + failures.size() + " error(s) enforcing stop policy on functions: " + + failures; + if (this.propagateStopFailures) { + throw new PulsarFunctionException(msg, failures); + } + this.logger.error(() -> msg); + } + } + catch (PulsarClientException ex) { + String msg = "Unable to enforce stop policy on functions - could not create PulsarAdmin: " + + ex.getMessage(); + if (this.propagateStopFailures) { + throw new PulsarException(msg, ex); + } + this.logger.error(ex, () -> msg); + } + } + + private Optional enforceStopPolicyOnFunction(PulsarFunctionOperations function, PulsarAdmin admin) { + return switch (function.stopPolicy()) { + case NONE -> { + this.logger.info(() -> String.format("No stop policy for %s - leaving alone", functionDesc(function))); + yield Optional.empty(); + } + case STOP -> { + this.logger.info(() -> String.format("Stopping %s", functionDesc(function))); + yield safeInvoke(() -> function.stop(admin)); + } + case DELETE -> { + this.logger.info(() -> String.format("Deleting %s", functionDesc(function))); + yield safeInvoke(() -> function.delete(admin)); + } + }; + } + + private Optional safeInvoke(Runnable invocation) { + try { + invocation.run(); + } + catch (Exception ex) { + return Optional.of(ex); + } + return Optional.empty(); + } + private String functionDesc(PulsarFunctionOperations function) { return String.format("'%s' %s", function.name(), function.type().toString().toLowerCase()); } 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 index ca03e4ed..b91597f4 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionOperations.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionOperations.java @@ -26,6 +26,7 @@ import org.apache.pulsar.common.io.SinkConfig; import org.apache.pulsar.common.io.SourceConfig; import org.springframework.core.log.LogAccessor; +import org.springframework.pulsar.PulsarException; /** * Provides operations for a particular function type. @@ -73,47 +74,16 @@ public interface PulsarFunctionOperations { 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 + * Gets the action to take on the function when the server is stopped. + * @return the function stop policy */ - 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; + FunctionStopPolicy stopPolicy(); /** * 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 + * @throws PulsarAdminException if anything goes wrong */ default boolean functionExists(PulsarAdmin admin) throws PulsarAdminException { return getIfExists(admin).isPresent(); @@ -136,6 +106,57 @@ public interface PulsarFunctionOperations { } } + /** + * 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; + + /** + * Creates the function using the file-based create api. + * @param admin the admin client + * @throws PulsarAdminException if anything goes wrong + */ + void create(PulsarAdmin admin) throws PulsarAdminException; + + /** + * Creates the function using the url-based create api. + * @param admin the admin client + * @throws PulsarAdminException if anything goes wrong + */ + void createWithUrl(PulsarAdmin admin) throws PulsarAdminException; + + /** + * Updates the function using the file-based update api. + * @param admin the admin client + * @throws PulsarAdminException if anything goes wrong + */ + void update(PulsarAdmin admin) throws PulsarAdminException; + + /** + * Updates the function using the url-based update api. + * @param admin the admin client + * @throws PulsarAdminException if anything goes wrong + */ + void updateWithUrl(PulsarAdmin admin) throws PulsarAdminException; + + /** + * Stops the function. + * @param admin the admin client + * @throws PulsarException if anything goes wrong + */ + void stop(PulsarAdmin admin); + + /** + * Deletes the function. + * @param admin the admin client + * @throws PulsarException if anything goes wrong + */ + void delete(PulsarAdmin admin); + /** * The type of function the operations handle. */ @@ -152,4 +173,20 @@ public interface PulsarFunctionOperations { } + /** + * The action to take on the function when the server stops. + */ + enum FunctionStopPolicy { + + /** Do nothing - leave the function alone. */ + NONE, + + /** Stop the function if running. */ + STOP, + + /** Delete the function. */ + DELETE + + } + } 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 index ac0b6375..186f138a 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSink.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSink.java @@ -23,16 +23,23 @@ import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.functions.UpdateOptions; import org.apache.pulsar.common.io.SinkConfig; +import org.springframework.pulsar.PulsarException; + /** * Represents a Pulsar Sink backed by a {@link SinkConfig}. * @param config the sink details + * @param stopPolicy the action to take on the sink when the server is stopped * @param updateOptions the options to use during an update operation (optional) * * @author Chris Bono */ -public record PulsarSink(SinkConfig config, +public record PulsarSink(SinkConfig config, FunctionStopPolicy stopPolicy, @Nullable UpdateOptions updateOptions) implements PulsarFunctionOperations { + public PulsarSink(SinkConfig config, @Nullable UpdateOptions updateOptions) { + this(config, FunctionStopPolicy.DELETE, updateOptions); + } + @Override public String name() { return config().getName(); @@ -72,4 +79,24 @@ public record PulsarSink(SinkConfig config, public void create(PulsarAdmin admin) throws PulsarAdminException { admin.sinks().createSink(config(), archive()); } + + @Override + public void stop(PulsarAdmin admin) { + try { + admin.sinks().stopSink(config().getTenant(), config().getNamespace(), config().getName()); + } + catch (PulsarAdminException e) { + throw new PulsarException(e.getMessage(), e); + } + } + + @Override + public void delete(PulsarAdmin admin) { + try { + admin.sinks().deleteSink(config().getTenant(), config().getNamespace(), config().getName()); + } + catch (PulsarAdminException e) { + throw new PulsarException(e.getMessage(), e); + } + } } 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 index 9b436076..f06e62b1 100644 --- a/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSource.java +++ b/spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSource.java @@ -23,16 +23,23 @@ import org.apache.pulsar.client.admin.PulsarAdminException; import org.apache.pulsar.common.functions.UpdateOptions; import org.apache.pulsar.common.io.SourceConfig; +import org.springframework.pulsar.PulsarException; + /** * Represents a Pulsar Source backed by a {@link SourceConfig}. * @param config the source details + * @param stopPolicy the action to take on the source when the server is stopped * @param updateOptions the options to use during an update operation (optional) * * @author Chris Bono */ -public record PulsarSource(SourceConfig config, +public record PulsarSource(SourceConfig config, FunctionStopPolicy stopPolicy, @Nullable UpdateOptions updateOptions) implements PulsarFunctionOperations { + public PulsarSource(SourceConfig config, @Nullable UpdateOptions updateOptions) { + this(config, FunctionStopPolicy.DELETE, updateOptions); + } + @Override public String name() { return config().getName(); @@ -72,4 +79,24 @@ public record PulsarSource(SourceConfig config, public void create(PulsarAdmin admin) throws PulsarAdminException { admin.sources().createSource(config(), archive()); } + + @Override + public void stop(PulsarAdmin admin) { + try { + admin.sources().stopSource(config().getTenant(), config().getNamespace(), config().getName()); + } + catch (PulsarAdminException e) { + throw new PulsarException(e.getMessage(), e); + } + } + + @Override + public void delete(PulsarAdmin admin) { + try { + admin.sources().deleteSource(config().getTenant(), config().getNamespace(), config().getName()); + } + catch (PulsarAdminException e) { + throw new PulsarException(e.getMessage(), e); + } + } } 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 index f6630267..91aecdd5 100644 --- a/spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionAdministrationTests.java +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionAdministrationTests.java @@ -20,6 +20,8 @@ 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.doThrow; +import static org.mockito.Mockito.inOrder; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.verify; @@ -29,17 +31,14 @@ 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.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.mockito.InOrder; import org.mockito.Mockito; import org.springframework.beans.factory.support.StaticListableBeanFactory; @@ -48,6 +47,7 @@ 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; +import org.springframework.pulsar.function.PulsarFunctionOperations.FunctionStopPolicy; /** * Tests for {@link PulsarFunctionAdministration}. @@ -56,278 +56,156 @@ import org.springframework.pulsar.function.PulsarFunctionAdministration.PulsarFu */ class PulsarFunctionAdministrationTests { - private PulsarAdmin pulsarAdmin = mock(PulsarAdmin.class, Mockito.RETURNS_DEEP_STUBS); + private PulsarAdmin pulsarAdmin; - private PulsarAdministration springPulsarAdmin = mock(PulsarAdministration.class); + private PulsarAdministration springPulsarAdmin; private PulsarFunctionAdministration functionAdmin; private StaticListableBeanFactory beanFactory; + private PulsarFunction function1; + + private PulsarSink sink1; + + private PulsarSource source1; + @BeforeEach - void setupAdminsAndBeanFactory() throws PulsarClientException { + void setupSharedMocks() throws PulsarClientException, PulsarAdminException { + pulsarAdmin = mock(PulsarAdmin.class, Mockito.RETURNS_DEEP_STUBS); + springPulsarAdmin = mock(PulsarAdministration.class); 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); + beanFactory.getBeanProvider(PulsarSource.class), true, true, true); + // create function, sink, source mock but do not add to bean factory + 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); } @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)); + class ProperCreateUpdateApiCalled { + @ParameterizedTest + @ValueSource(strings = { "myfunc.jar", "builtin://myfunc" }) + void create(String archive) throws PulsarAdminException { + PulsarFunction function = setupMockFunction(archive); + when(function.functionExists(pulsarAdmin)).thenReturn(false); functionAdmin.createOrUpdateUserDefinedFunctions(); - - verify(pulsarAdmin.functions()).getFunction(functionConfig.getTenant(), functionConfig.getNamespace(), - functionConfig.getName()); - verify(pulsarAdmin.functions()).createFunction(functionConfig, functionConfig.getJar()); + verify(function).create(pulsarAdmin); } - @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)); - + @ParameterizedTest + @ValueSource(strings = { "https://myfunc.jar", "file:///myfunc.jar", "function://myfunc.jar", + "sink://myfunc.jar", "source://myfunc.jar" }) + void createWithUrl(String archive) throws PulsarAdminException { + PulsarFunction function = setupMockFunction(archive); + when(function.functionExists(pulsarAdmin)).thenReturn(false); functionAdmin.createOrUpdateUserDefinedFunctions(); + verify(function).createWithUrl(pulsarAdmin); + } - verify(pulsarAdmin.functions()).getFunction(functionConfigNew.getTenant(), functionConfigNew.getNamespace(), - functionConfigNew.getName()); - verify(pulsarAdmin.functions()).updateFunction(functionConfigNew, functionConfigNew.getJar(), null); + @ParameterizedTest + @ValueSource(strings = { "myfunc.jar", "builtin://myfunc" }) + void update(String archive) throws PulsarAdminException { + PulsarFunction function = setupMockFunction(archive); + when(function.functionExists(pulsarAdmin)).thenReturn(true); + functionAdmin.createOrUpdateUserDefinedFunctions(); + verify(function).update(pulsarAdmin); + } + + @ParameterizedTest + @ValueSource(strings = { "https://myfunc.jar", "file:///myfunc.jar", "function://myfunc.jar", + "sink://myfunc.jar", "source://myfunc.jar" }) + void updateWithUrl(String archive) throws PulsarAdminException { + PulsarFunction function = setupMockFunction(archive); + when(function.functionExists(pulsarAdmin)).thenReturn(true); + functionAdmin.createOrUpdateUserDefinedFunctions(); + verify(function).updateWithUrl(pulsarAdmin); + } + + private PulsarFunction setupMockFunction(String archive) { + PulsarFunction function = mock(PulsarFunction.class); + when(function.name()).thenReturn("function1"); + when(function.archive()).thenReturn(archive); + beanFactory.addBean("myFunction", function); + return function; } } @Nested - class ManagePulsarSinks { - - private SinkConfig sinkConfig = SinkConfig.builder().tenant("tenant1").namespace("namespace1").name("sink1") - .archive("sink1.jar").build(); + class ProperCreateUpdateProcessOrder { @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)); - + void noFunctionsProvided() throws PulsarClientException { functionAdmin.createOrUpdateUserDefinedFunctions(); - - verify(pulsarAdmin.sinks()).getSink(sinkConfig.getTenant(), sinkConfig.getNamespace(), - sinkConfig.getName()); - verify(pulsarAdmin.sinks()).createSink(sinkConfig, sinkConfig.getArchive()); + verify(springPulsarAdmin, never()).createAdminClient(); + verifyNoInteractions(pulsarAdmin); + assertThat(functionAdmin.getProcessedFunctions()).isEmpty(); } @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)); - + void functionProvided() { + beanFactory.addBean("function1", function1); 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()); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(function1); } @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)); - + void functionAndSinkProvided() { + beanFactory.addBean("function1", function1); + beanFactory.addBean("sink1", sink1); functionAdmin.createOrUpdateUserDefinedFunctions(); - - verify(pulsarAdmin.sources()).getSource(sourceConfigNew.getTenant(), sourceConfigNew.getNamespace(), - sourceConfigNew.getName()); - verify(pulsarAdmin.sources()).updateSource(sourceConfigNew, sourceConfigNew.getArchive(), updateOptions); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(function1, sink1); } - } - - @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); - + @Test + void functionSinkAndSourceProvided() { 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); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(function1, sink1, source1); } @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); + void sinkProvided() { + beanFactory.addBean("sink1", sink1); + functionAdmin.createOrUpdateUserDefinedFunctions(); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(sink1); } @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); + void sinkAndSourceProvided() { + beanFactory.addBean("sink1", sink1); + beanFactory.addBean("source1", source1); + functionAdmin.createOrUpdateUserDefinedFunctions(); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(sink1, 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); + void sourceProvided() { + beanFactory.addBean("source1", source1); + functionAdmin.createOrUpdateUserDefinedFunctions(); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(source1); } - @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); - } + @Nested + class ProperCreateUpdateErrorHandling { - @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); + @BeforeEach + void provideFunctionsToBeanFactory() { + beanFactory.addBean("function1", function1); + beanFactory.addBean("sink1", sink1); + beanFactory.addBean("source1", source1); } @Test @@ -338,37 +216,326 @@ class PulsarFunctionAdministrationTests { .hasMessageContaining("Unable to create/update functions - could not create PulsarAdmin: NOPE"); } + @Nested + class WithFailFast { + + @Test + void firstProcessedFunctionFails() 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); + assertThat(functionAdmin.getProcessedFunctions()).isEmpty(); + } + + @Test + void middleProcessedFunctionFails() 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); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(function1); + } + + @Test + void lastProcessedFunctionFails() 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); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(function1, sink1); + } + + } + + @Nested + class WithoutFailFast { + + @BeforeEach + void disableFailFastOnFunctionAdmin() { + functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, + beanFactory.getBeanProvider(PulsarFunction.class), + beanFactory.getBeanProvider(PulsarSink.class), beanFactory.getBeanProvider(PulsarSource.class), + false, true, true); + } + + @Test + void firstProcessedFunctionFails() 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); + verify(sink1).create(pulsarAdmin); + verify(source1).create(pulsarAdmin); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(sink1, source1); + } + + @Test + void middleProcessedFunctionFails() 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); + verify(source1).create(pulsarAdmin); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(function1, source1); + } + + @Test + void lastProcessedFunctionFails() 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); + assertThat(functionAdmin.getProcessedFunctions()).containsExactly(function1, sink1); + } + + @Test + void allProcessedFunctionsFail() throws PulsarAdminException { + 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); + assertThat(functionAdmin.getProcessedFunctions()).isEmpty(); + } + + } + + @Nested + @ExtendWith(OutputCaptureExtension.class) + class WithPropagationDisabled { + + @BeforeEach + void disablePropagationOnFunctionAdmin() { + functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, + beanFactory.getBeanProvider(PulsarFunction.class), + beanFactory.getBeanProvider(PulsarSink.class), beanFactory.getBeanProvider(PulsarSource.class), + true, false, false); + } + + @Test + void createAdminClientFails(CapturedOutput output) throws PulsarClientException { + beanFactory.addBean("function1", function1); + 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 { + beanFactory.addBean("function1", function1); + when(function1.functionExists(pulsarAdmin)).thenThrow(new PulsarAdminException("BOOM")); + functionAdmin.createOrUpdateUserDefinedFunctions(); + assertThat(output).contains("Encountered 1 error(s) creating/updating functions:", + "PulsarAdminException: BOOM"); + } + + } + } @Nested - @ExtendWith(OutputCaptureExtension.class) - class ProcessHandlingPropagationDisabled { + class ProperStopPolicyApiCalled { + + @Test + void none() { + PulsarFunction function = mock(PulsarFunction.class); + when(function.stopPolicy()).thenReturn(FunctionStopPolicy.NONE); + functionAdmin.getProcessedFunctions().add(function); + functionAdmin.enforceStopPolicyOnUserDefinedFunctions(); + verify(function).stopPolicy(); + verifyNoMoreInteractions(function); + } + + @Test + void stop() { + PulsarFunction function = mock(PulsarFunction.class); + when(function.stopPolicy()).thenReturn(FunctionStopPolicy.STOP); + functionAdmin.getProcessedFunctions().add(function); + functionAdmin.enforceStopPolicyOnUserDefinedFunctions(); + verify(function).stop(pulsarAdmin); + } + + @Test + void delete() { + PulsarFunction function = mock(PulsarFunction.class); + when(function.stopPolicy()).thenReturn(FunctionStopPolicy.DELETE); + functionAdmin.getProcessedFunctions().add(function); + functionAdmin.enforceStopPolicyOnUserDefinedFunctions(); + verify(function).delete(pulsarAdmin); + } + + } + + @Nested + class ProperStopPolicyProcessOrder { @BeforeEach - void setupFunctionAdminWithPropagationDisabled() { - functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, - beanFactory.getBeanProvider(PulsarFunction.class), beanFactory.getBeanProvider(PulsarSink.class), - beanFactory.getBeanProvider(PulsarSource.class), true, false); + void setStopPolicyOnFunctions() { + when(function1.stopPolicy()).thenReturn(FunctionStopPolicy.STOP); + when(sink1.stopPolicy()).thenReturn(FunctionStopPolicy.STOP); + when(source1.stopPolicy()).thenReturn(FunctionStopPolicy.STOP); } @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"); + void noFunctionsProvided() throws PulsarClientException { + functionAdmin.enforceStopPolicyOnUserDefinedFunctions(); + verify(springPulsarAdmin, never()).createAdminClient(); + verifyNoInteractions(pulsarAdmin); } @Test - void processedFunctionFails(CapturedOutput output) throws PulsarAdminException { - PulsarFunction function1 = mock(PulsarFunction.class); + void processedInReverseOrder() { beanFactory.addBean("function1", function1); - PulsarAdminException ex = new PulsarAdminException("BOOM"); - when(function1.functionExists(pulsarAdmin)).thenThrow(ex); + beanFactory.addBean("sink1", sink1); + beanFactory.addBean("source1", source1); + functionAdmin.getProcessedFunctions().add(function1); + functionAdmin.getProcessedFunctions().add(sink1); + functionAdmin.getProcessedFunctions().add(source1); + functionAdmin.enforceStopPolicyOnUserDefinedFunctions(); + InOrder inOrder = inOrder(function1, sink1, source1); + inOrder.verify(source1).stop(pulsarAdmin); + inOrder.verify(sink1).stop(pulsarAdmin); + inOrder.verify(function1).stop(pulsarAdmin); + } - functionAdmin.createOrUpdateUserDefinedFunctions(); + } + + @Nested + class ProperStopPolicyErrorHandling { + + @BeforeEach + void setStopPolicyOnFunctionsAndAddToProcessedList() { + when(function1.stopPolicy()).thenReturn(FunctionStopPolicy.STOP); + when(sink1.stopPolicy()).thenReturn(FunctionStopPolicy.STOP); + when(source1.stopPolicy()).thenReturn(FunctionStopPolicy.STOP); + functionAdmin.getProcessedFunctions().add(function1); + functionAdmin.getProcessedFunctions().add(sink1); + functionAdmin.getProcessedFunctions().add(source1); + } + + @Test + void createAdminClientFails() throws PulsarClientException { + when(springPulsarAdmin.createAdminClient()).thenThrow(new PulsarClientException("NOPE")); + assertThatThrownBy(() -> functionAdmin.enforceStopPolicyOnUserDefinedFunctions()) + .isInstanceOf(PulsarException.class).hasMessageContaining( + "Unable to enforce stop policy on functions - could not create PulsarAdmin: NOPE"); + } + + @Test + void firstProcessedFunctionFails() { + PulsarException ex = new PulsarException("BOOM"); + doThrow(ex).when(source1).stop(pulsarAdmin); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.enforceStopPolicyOnUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(source1, ex)); + verify(sink1).stop(pulsarAdmin); + verify(function1).stop(pulsarAdmin); + } + + @Test + void middleProcessedFunctionFails() { + PulsarException ex = new PulsarException("BOOM"); + doThrow(ex).when(sink1).stop(pulsarAdmin); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.enforceStopPolicyOnUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(sink1, ex)); + verify(source1).stop(pulsarAdmin); + verify(function1).stop(pulsarAdmin); + } + + @Test + void lastProcessedFunctionFails() { + PulsarException ex = new PulsarException("BOOM"); + doThrow(ex).when(function1).stop(pulsarAdmin); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.enforceStopPolicyOnUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(function1, ex)); + verify(source1).stop(pulsarAdmin); + verify(sink1).stop(pulsarAdmin); + } + + @Test + void allProcessedFunctionsFail() { + PulsarException ex1 = new PulsarException("BOOM1"); + PulsarException ex2 = new PulsarException("BOOM2"); + PulsarException ex3 = new PulsarException("BOOM3"); + doThrow(ex1).when(source1).stop(pulsarAdmin); + doThrow(ex2).when(sink1).stop(pulsarAdmin); + doThrow(ex3).when(function1).stop(pulsarAdmin); + PulsarFunctionException thrown = catchThrowableOfType( + () -> functionAdmin.enforceStopPolicyOnUserDefinedFunctions(), PulsarFunctionException.class); + assertThat(thrown.getFailures()).containsExactly(entry(source1, ex1), entry(sink1, ex2), + entry(function1, ex3)); + } + + @Nested + @ExtendWith(OutputCaptureExtension.class) + class WithPropagationDisabled { + + @BeforeEach + void disableStopPropagationOnFunctionAdmin() { + functionAdmin = new PulsarFunctionAdministration(springPulsarAdmin, + beanFactory.getBeanProvider(PulsarFunction.class), + beanFactory.getBeanProvider(PulsarSink.class), beanFactory.getBeanProvider(PulsarSource.class), + true, false, false); + } + + @Test + void createAdminClientFails(CapturedOutput output) throws PulsarClientException { + functionAdmin.getProcessedFunctions().add(function1); + when(springPulsarAdmin.createAdminClient()).thenThrow(new PulsarClientException("NOPE")); + functionAdmin.enforceStopPolicyOnUserDefinedFunctions(); + assertThat(output) + .contains("Unable to enforce stop policy on functions - could not create PulsarAdmin: NOPE"); + } + + @Test + void processedFunctionFails(CapturedOutput output) { + functionAdmin.getProcessedFunctions().add(function1); + doThrow(new PulsarException("BOOM")).when(function1).stop(pulsarAdmin); + functionAdmin.enforceStopPolicyOnUserDefinedFunctions(); + assertThat(output).contains("Encountered 1 error(s) enforcing stop policy on functions:", + "PulsarException: BOOM"); + } - assertThat(output).contains("Encountered 1 error(s) creating/updating functions:", - "PulsarAdminException: BOOM"); } } diff --git a/spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionsTests.java b/spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionsTests.java new file mode 100644 index 00000000..4901332b --- /dev/null +++ b/spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionsTests.java @@ -0,0 +1,382 @@ +/* + * 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.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import org.apache.pulsar.client.admin.Functions; +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.admin.Sinks; +import org.apache.pulsar.client.admin.Sources; +import org.apache.pulsar.common.functions.FunctionConfig; +import org.apache.pulsar.common.io.SinkConfig; +import org.apache.pulsar.common.io.SourceConfig; +import org.junit.jupiter.api.Nested; +import org.junit.jupiter.api.Test; +import org.mockito.Mockito; + +import org.springframework.pulsar.PulsarException; +import org.springframework.pulsar.function.PulsarFunctionOperations.FunctionStopPolicy; +import org.springframework.pulsar.function.PulsarFunctionOperations.FunctionType; + +/** + * Tests for all "Pulsar Functions" (ie. {@link PulsarFunction}, + * {@link PulsarSink}, and {@link PulsarSource}). + * + * @author Chris Bono + */ +class PulsarFunctionsTests { + + private static final String TENANT = "tenant1"; + + private static final String NAMESPACE = "namespace1"; + + private PulsarAdmin pulsarAdmin = mock(PulsarAdmin.class, Mockito.RETURNS_DEEP_STUBS); + + @Nested + class PulsarFunctionApi { + + private static final String NAME = "function1"; + + private static final String JAR = "function1.jar"; + + private FunctionConfig functionConfig = FunctionConfig.builder().tenant(TENANT).namespace(NAMESPACE).name(NAME) + .jar(JAR).build(); + + private PulsarFunction function = new PulsarFunction(functionConfig, null); + + @Test + void accessors() { + PulsarFunction function = new PulsarFunction(functionConfig, FunctionStopPolicy.STOP, null); + assertThat(function.config()).isEqualTo(functionConfig); + assertThat(function.name()).isEqualTo(NAME); + assertThat(function.type()).isEqualTo(FunctionType.FUNCTION); + assertThat(function.archive()).isEqualTo(JAR); + assertThat(function.stopPolicy()).isEqualTo(FunctionStopPolicy.STOP); + } + + @Test + void defaultStopPolicy() { + assertThat(function.stopPolicy()).isEqualTo(FunctionStopPolicy.DELETE); + } + + @Test + void get() throws PulsarAdminException { + when(pulsarAdmin.functions().getFunction(anyString(), anyString(), anyString())).thenReturn(functionConfig); + assertThat(function.get(pulsarAdmin)).isSameAs(functionConfig); + verify(pulsarAdmin.functions()).getFunction(TENANT, NAMESPACE, NAME); + } + + @Test + void getIfExistsWithExistingFunction() throws PulsarAdminException { + when(pulsarAdmin.functions().getFunction(anyString(), anyString(), anyString())).thenReturn(functionConfig); + assertThat(function.getIfExists(pulsarAdmin)).hasValue(functionConfig); + assertThat(function.functionExists(pulsarAdmin)).isTrue(); + } + + @Test + void getIfExistsWithNonExistentFunction() throws PulsarAdminException { + when(pulsarAdmin.functions().getFunction(anyString(), anyString(), anyString())) + .thenThrow(new NotFoundException(null, "400", 400)); + assertThat(function.getIfExists(pulsarAdmin)).isEmpty(); + assertThat(function.functionExists(pulsarAdmin)).isFalse(); + } + + @Test + void create() throws PulsarAdminException { + function.create(pulsarAdmin); + verify(pulsarAdmin.functions()).createFunction(functionConfig, JAR); + } + + @Test + void createWithUrl() throws PulsarAdminException { + function.createWithUrl(pulsarAdmin); + verify(pulsarAdmin.functions()).createFunctionWithUrl(functionConfig, JAR); + } + + @Test + void update() throws PulsarAdminException { + function.update(pulsarAdmin); + verify(pulsarAdmin.functions()).updateFunction(functionConfig, JAR, null); + } + + @Test + void updateWithUrl() throws PulsarAdminException { + function.updateWithUrl(pulsarAdmin); + verify(pulsarAdmin.functions()).updateFunctionWithUrl(functionConfig, JAR, null); + } + + @Test + void stop() throws PulsarAdminException { + function.stop(pulsarAdmin); + verify(pulsarAdmin.functions()).stopFunction(TENANT, NAMESPACE, NAME); + } + + @Test + void stopWrapsPulsarAdminException() throws PulsarAdminException { + Functions functions = mock(Functions.class); + when(pulsarAdmin.functions()).thenReturn(functions); + PulsarAdminException paex = new PulsarAdminException("bad-stop"); + doThrow(paex).when(functions).stopFunction(anyString(), anyString(), anyString()); + assertThatThrownBy(() -> function.stop(pulsarAdmin)).isInstanceOf(PulsarException.class) + .hasMessageContaining("bad-stop").hasCause(paex); + } + + @Test + void delete() throws PulsarAdminException { + function.delete(pulsarAdmin); + verify(pulsarAdmin.functions()).deleteFunction(TENANT, NAMESPACE, NAME); + } + + @Test + void deleteWrapsPulsarAdminException() throws PulsarAdminException { + PulsarFunction function = new PulsarFunction(functionConfig, null); + Functions functions = mock(Functions.class); + when(pulsarAdmin.functions()).thenReturn(functions); + PulsarAdminException paex = new PulsarAdminException("bad-delete"); + doThrow(paex).when(functions).deleteFunction(anyString(), anyString(), anyString()); + assertThatThrownBy(() -> function.delete(pulsarAdmin)).isInstanceOf(PulsarException.class) + .hasMessageContaining("bad-delete").hasCause(paex); + } + + } + + @Nested + class PulsarSinkApi { + + private static final String NAME = "sink1"; + + private static final String JAR = "sink1.jar"; + + private SinkConfig sinkConfig = SinkConfig.builder().tenant(TENANT).namespace(NAMESPACE).name(NAME).archive(JAR) + .build(); + + private PulsarSink sink = new PulsarSink(sinkConfig, null); + + @Test + void accessors() { + PulsarSink sink = new PulsarSink(sinkConfig, FunctionStopPolicy.STOP, null); + assertThat(sink.config()).isEqualTo(sinkConfig); + assertThat(sink.name()).isEqualTo(NAME); + assertThat(sink.type()).isEqualTo(FunctionType.SINK); + assertThat(sink.archive()).isEqualTo(JAR); + assertThat(sink.stopPolicy()).isEqualTo(FunctionStopPolicy.STOP); + } + + @Test + void defaultStopPolicy() { + assertThat(sink.stopPolicy()).isEqualTo(FunctionStopPolicy.DELETE); + } + + @Test + void get() throws PulsarAdminException { + when(pulsarAdmin.sinks().getSink(anyString(), anyString(), anyString())).thenReturn(sinkConfig); + assertThat(sink.get(pulsarAdmin)).isSameAs(sinkConfig); + verify(pulsarAdmin.sinks()).getSink(TENANT, NAMESPACE, NAME); + } + + @Test + void getIfExistsWithExistingSink() throws PulsarAdminException { + when(pulsarAdmin.sinks().getSink(anyString(), anyString(), anyString())).thenReturn(sinkConfig); + assertThat(sink.getIfExists(pulsarAdmin)).hasValue(sinkConfig); + assertThat(sink.functionExists(pulsarAdmin)).isTrue(); + } + + @Test + void getIfExistsWithNonExistentSink() throws PulsarAdminException { + when(pulsarAdmin.sinks().getSink(anyString(), anyString(), anyString())) + .thenThrow(new NotFoundException(null, "400", 400)); + assertThat(sink.getIfExists(pulsarAdmin)).isEmpty(); + assertThat(sink.functionExists(pulsarAdmin)).isFalse(); + } + + @Test + void create() throws PulsarAdminException { + sink.create(pulsarAdmin); + verify(pulsarAdmin.sinks()).createSink(sinkConfig, JAR); + } + + @Test + void createWithUrl() throws PulsarAdminException { + sink.createWithUrl(pulsarAdmin); + verify(pulsarAdmin.sinks()).createSinkWithUrl(sinkConfig, JAR); + } + + @Test + void update() throws PulsarAdminException { + sink.update(pulsarAdmin); + verify(pulsarAdmin.sinks()).updateSink(sinkConfig, JAR, null); + } + + @Test + void updateWithUrl() throws PulsarAdminException { + sink.updateWithUrl(pulsarAdmin); + verify(pulsarAdmin.sinks()).updateSinkWithUrl(sinkConfig, JAR, null); + } + + @Test + void stop() throws PulsarAdminException { + sink.stop(pulsarAdmin); + verify(pulsarAdmin.sinks()).stopSink(TENANT, NAMESPACE, NAME); + } + + @Test + void stopWrapsPulsarAdminException() throws PulsarAdminException { + Sinks sinks = mock(Sinks.class); + when(pulsarAdmin.sinks()).thenReturn(sinks); + PulsarAdminException paex = new PulsarAdminException("bad-stop"); + doThrow(paex).when(sinks).stopSink(anyString(), anyString(), anyString()); + assertThatThrownBy(() -> sink.stop(pulsarAdmin)).isInstanceOf(PulsarException.class) + .hasMessageContaining("bad-stop").hasCause(paex); + } + + @Test + void delete() throws PulsarAdminException { + sink.delete(pulsarAdmin); + verify(pulsarAdmin.sinks()).deleteSink(TENANT, NAMESPACE, NAME); + } + + @Test + void deleteWrapsPulsarAdminException() throws PulsarAdminException { + PulsarSink sink = new PulsarSink(sinkConfig, null); + Sinks sinks = mock(Sinks.class); + when(pulsarAdmin.sinks()).thenReturn(sinks); + PulsarAdminException paex = new PulsarAdminException("bad-delete"); + doThrow(paex).when(sinks).deleteSink(anyString(), anyString(), anyString()); + assertThatThrownBy(() -> sink.delete(pulsarAdmin)).isInstanceOf(PulsarException.class) + .hasMessageContaining("bad-delete").hasCause(paex); + } + + } + + @Nested + class PulsarSourceApi { + + private static final String NAME = "source1"; + + private static final String JAR = "source1.jar"; + + private SourceConfig sourceConfig = SourceConfig.builder().tenant(TENANT).namespace(NAMESPACE).name(NAME) + .archive(JAR).build(); + + private PulsarSource source = new PulsarSource(sourceConfig, null); + + @Test + void accessors() { + PulsarSource source = new PulsarSource(sourceConfig, FunctionStopPolicy.STOP, null); + assertThat(source.config()).isEqualTo(sourceConfig); + assertThat(source.name()).isEqualTo(NAME); + assertThat(source.type()).isEqualTo(FunctionType.SOURCE); + assertThat(source.archive()).isEqualTo(JAR); + assertThat(source.stopPolicy()).isEqualTo(FunctionStopPolicy.STOP); + } + + @Test + void defaultStopPolicy() { + assertThat(source.stopPolicy()).isEqualTo(FunctionStopPolicy.DELETE); + } + + @Test + void get() throws PulsarAdminException { + when(pulsarAdmin.sources().getSource(anyString(), anyString(), anyString())).thenReturn(sourceConfig); + assertThat(source.get(pulsarAdmin)).isSameAs(sourceConfig); + verify(pulsarAdmin.sources()).getSource(TENANT, NAMESPACE, NAME); + } + + @Test + void getIfExistsWithExistingSource() throws PulsarAdminException { + when(pulsarAdmin.sources().getSource(anyString(), anyString(), anyString())).thenReturn(sourceConfig); + assertThat(source.getIfExists(pulsarAdmin)).hasValue(sourceConfig); + assertThat(source.functionExists(pulsarAdmin)).isTrue(); + } + + @Test + void getIfExistsWithNonExistentSource() throws PulsarAdminException { + when(pulsarAdmin.sources().getSource(anyString(), anyString(), anyString())) + .thenThrow(new NotFoundException(null, "400", 400)); + assertThat(source.getIfExists(pulsarAdmin)).isEmpty(); + assertThat(source.functionExists(pulsarAdmin)).isFalse(); + } + + @Test + void create() throws PulsarAdminException { + source.create(pulsarAdmin); + verify(pulsarAdmin.sources()).createSource(sourceConfig, JAR); + } + + @Test + void createWithUrl() throws PulsarAdminException { + source.createWithUrl(pulsarAdmin); + verify(pulsarAdmin.sources()).createSourceWithUrl(sourceConfig, JAR); + } + + @Test + void update() throws PulsarAdminException { + source.update(pulsarAdmin); + verify(pulsarAdmin.sources()).updateSource(sourceConfig, JAR, null); + } + + @Test + void updateWithUrl() throws PulsarAdminException { + source.updateWithUrl(pulsarAdmin); + verify(pulsarAdmin.sources()).updateSourceWithUrl(sourceConfig, JAR, null); + } + + @Test + void stop() throws PulsarAdminException { + source.stop(pulsarAdmin); + verify(pulsarAdmin.sources()).stopSource(TENANT, NAMESPACE, NAME); + } + + @Test + void stopWrapsPulsarAdminException() throws PulsarAdminException { + Sources sources = mock(Sources.class); + when(pulsarAdmin.sources()).thenReturn(sources); + PulsarAdminException paex = new PulsarAdminException("bad-stop"); + doThrow(paex).when(sources).stopSource(anyString(), anyString(), anyString()); + assertThatThrownBy(() -> source.stop(pulsarAdmin)).isInstanceOf(PulsarException.class) + .hasMessageContaining("bad-stop").hasCause(paex); + } + + @Test + void delete() throws PulsarAdminException { + source.delete(pulsarAdmin); + verify(pulsarAdmin.sources()).deleteSource(TENANT, NAMESPACE, NAME); + } + + @Test + void deleteWrapsPulsarAdminException() throws PulsarAdminException { + PulsarSource source = new PulsarSource(sourceConfig, null); + Sources sources = mock(Sources.class); + when(pulsarAdmin.sources()).thenReturn(sources); + PulsarAdminException paex = new PulsarAdminException("bad-delete"); + doThrow(paex).when(sources).deleteSource(anyString(), anyString(), anyString()); + assertThatThrownBy(() -> source.delete(pulsarAdmin)).isInstanceOf(PulsarException.class) + .hasMessageContaining("bad-delete").hasCause(paex); + } + + } + +}