Spring/spring-pulsar
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add stop policy for Pulsar Functions (#271)

Browse Source
This commit is contained in:
Chris Bono
2023-01-13 16:04:52 -06:00
committed by GitHub
parent 195b2fb77b
commit 66beedef86
14 changed files with 1137 additions and 308 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
spring-pulsar-docs/src/main/asciidoc/pulsar-function.adoc
View File

@@ -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 <<pulsar-admin.adoc#pulsar-admin-client,Pulsar Admin Client>> for available client options (including authentication). Other available application properties can be found in the <<application-properties.adoc#appendix.application-properties.pulsar-function,Appendix>> 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

7
spring-pulsar-sample-apps/sample-pulsar-functions/sample-signup-app/src/main/java/org/springframework/pulsar/sample/signup/config/AppConfig.java
View File

@@ -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);
}
}

3
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfiguration.java
View File

@@ -124,7 +124,8 @@ public class PulsarAutoConfiguration {
ObjectProvider<PulsarFunction> pulsarFunctions, ObjectProvider<PulsarSink> pulsarSinks,
ObjectProvider<PulsarSource> 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());
}
}

17
spring-pulsar-spring-boot-autoconfigure/src/main/java/org/springframework/pulsar/autoconfigure/PulsarProperties.java
View File

@@ -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 {

11
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarAutoConfigurationTests.java
View File

@@ -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

13
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarFunctionTests.java
View File

@@ -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<PulsarFunctionOperations<?>, Exception> failures = thrown.getFailures();
ApplicationContextException.class);
assertThat(thrown).hasCauseInstanceOf(PulsarFunctionException.class);
PulsarFunctionException cause = (PulsarFunctionException) thrown.getCause();
Map<PulsarFunctionOperations<?>, Exception> failures = cause.getFailures();
assertThat(failures).hasSize(1);
Map.Entry<PulsarFunctionOperations<?>, 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);
}
}

3
spring-pulsar-spring-boot-autoconfigure/src/test/java/org/springframework/pulsar/autoconfigure/PulsarPropertiesTests.java
View File

@@ -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();
}
}

29
spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunction.java
View File

@@ -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<FunctionConfig> {
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);
}
}
}

162
spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionAdministration.java
View File

@@ -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<PulsarSource> pulsarSources;
private final List<PulsarFunctionOperations<?>> 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<PulsarFunction> pulsarFunctions, ObjectProvider<PulsarSink> pulsarSinks,
ObjectProvider<PulsarSource> pulsarSources, boolean failFast, boolean propagateFailures) {
ObjectProvider<PulsarSource> 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.
*
* <p>
* 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<PulsarFunctionOperations<?>> allFunctions = Stream.concat(
Stream.concat(this.pulsarFunctions.orderedStream(), this.pulsarSinks.orderedStream()),
this.pulsarSources.orderedStream());
// Concat the functions/sinks/sources into a single stream
Stream<PulsarFunctionOperations<?>> allFunctions = Stream.concat(
Stream.concat(this.pulsarFunctions.orderedStream(), this.pulsarSinks.orderedStream()),
this.pulsarSources.orderedStream());
List<PulsarFunctionOperations<?>> 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<PulsarFunctionOperations<?>, Exception> failures = new LinkedHashMap<>();
for (PulsarFunctionOperations<?> function : allFunctions.toList()) {
for (PulsarFunctionOperations<?> function : functionsToProcess) {
Optional<Exception> 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<PulsarFunctionOperations<?>> getProcessedFunctions() {
return this.processedFunctions;
}
/**
* Called during server shutdown, enforces the stop policy on all Pulsar functions
* that were successfully processed during server startup.
*
* <p>
* The functions, sinks, and sources are processed in reverse startup order as
* follows:
* <ul>
* <li>The stop policy of each function is used to determine if the function should be
* stopped, removed, or left alone.
* </ul>
*
* <p>
* 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<PulsarFunctionOperations<?>, Exception> failures = new LinkedHashMap<>();
// Spin through the processed functions in reverse startup order
Collections.reverse(this.processedFunctions);
for (PulsarFunctionOperations<?> function : this.processedFunctions) {
Optional<Exception> 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<Exception> 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<Exception> 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());
}

107
spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarFunctionOperations.java
View File

@@ -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<T> {
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<T> {
}
}
/**
* 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<T> {
}
/**
* 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
}
}

29
spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSink.java
View File

@@ -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<SinkConfig> {
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);
}
}
}

29
spring-pulsar/src/main/java/org/springframework/pulsar/function/PulsarSource.java
View File

@@ -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<SourceConfig> {
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);
}
}
}

647
spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionAdministrationTests.java
View File

@@ -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");
}
}

382
spring-pulsar/src/test/java/org/springframework/pulsar/function/PulsarFunctionsTests.java Normal file
View File

@@ -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 &quot;Pulsar Functions&quot; (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);
}
}
}