From 1330de11c6126ffd62a07a949baf5722589a2703 Mon Sep 17 00:00:00 2001 From: David Turanski Date: Tue, 17 Aug 2021 10:20:53 -0400 Subject: [PATCH] Remove tasklauncher-function and tasklauncher-sink --- applications/sink/pom.xml | 1 - .../sink/tasklauncher-sink/README.adoc | 121 +------ applications/sink/tasklauncher-sink/pom.xml | 101 ------ .../sink/LaunchRequestConsumer.java | 210 ------------ .../app/tasklauncher/sink/PollingSink.java | 33 -- .../sink/TaskLauncherSinkConfiguration.java | 99 ------ .../tasklauncher/sink/TriggerProperties.java | 78 ----- ...onfiguration-metadata-whitelist.properties | 9 - ...dataflow-configuration-metadata.properties | 9 - .../sink/TaskLauncherSinkTests.java | 312 ------------------ .../source/time/KafkaTimeSourceTests.java | 1 - functions/function-dependencies/pom.xml | 5 - .../tasklauncher-function/README.adoc | 40 +-- .../function/tasklauncher-function/pom.xml | 41 --- .../cloud/fn/tasklauncher/LaunchRequest.java | 66 ---- .../fn/tasklauncher/TaskLauncherFunction.java | 167 ---------- .../TaskLauncherFunctionConfiguration.java | 46 --- .../TaskLauncherFunctionProperties.java | 38 --- .../TaskLauncherFunctionApplicationTests.java | 151 --------- functions/pom.xml | 1 - .../stream-applications-descriptor/pom.xml | 2 - .../META-INF/kafka-apps-docker.properties | 1 - .../META-INF/kafka-apps-harbor.properties | 1 - .../kafka-apps-maven-repo-url.properties | 2 - .../META-INF/kafka-apps-maven.properties | 2 - .../META-INF/rabbit-apps-docker.properties | 1 - .../META-INF/rabbit-apps-harbor.properties | 1 - .../rabbit-apps-maven-repo-url.properties | 2 - .../META-INF/rabbit-apps-maven.properties | 2 - 29 files changed, 9 insertions(+), 1534 deletions(-) delete mode 100644 applications/sink/tasklauncher-sink/pom.xml delete mode 100644 applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/LaunchRequestConsumer.java delete mode 100644 applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/PollingSink.java delete mode 100644 applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkConfiguration.java delete mode 100644 applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TriggerProperties.java delete mode 100644 applications/sink/tasklauncher-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties delete mode 100644 applications/sink/tasklauncher-sink/src/main/resources/META-INF/dataflow-configuration-metadata.properties delete mode 100644 applications/sink/tasklauncher-sink/src/test/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkTests.java delete mode 100644 functions/function/tasklauncher-function/pom.xml delete mode 100644 functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java delete mode 100644 functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunction.java delete mode 100644 functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionConfiguration.java delete mode 100644 functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java delete mode 100644 functions/function/tasklauncher-function/src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java diff --git a/applications/sink/pom.xml b/applications/sink/pom.xml index f505d495..d33a05f5 100644 --- a/applications/sink/pom.xml +++ b/applications/sink/pom.xml @@ -23,7 +23,6 @@ rabbit-sink router-sink sftp-sink - tasklauncher-sink s3-sink tcp-sink throughput-sink diff --git a/applications/sink/tasklauncher-sink/README.adoc b/applications/sink/tasklauncher-sink/README.adoc index b5a14ffe..276c09b7 100644 --- a/applications/sink/tasklauncher-sink/README.adoc +++ b/applications/sink/tasklauncher-sink/README.adoc @@ -1,119 +1,10 @@ -//tag::ref-doc[] = Tasklauncher Sink -This module consumes LaunchRequest messages from a `PollableMessageSource` and uses the Data Flow REST client to launch a registered task on a configured https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-tasks[task platform]. -The client must be configured to connect to a remote Data Flow Server, including any required authentication (see Configuration Options below). +[NOTE] +This module has been relocated to the https://github.com/spring-cloud/spring-cloud-dataflow/tree/main/spring-cloud-dataflow-tasklauncher/spring-cloud-dataflow-tasklauncher-sink[spring-cloud-data-flow repository]. +As of SCDF 2.9.x, this app is included in the SCDF build to ensure version compatibility with the SCDF server going forward. -This application launches a registered task definition using the Data Flow Server https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#api-guide-resources-task-executions-launching[REST API]. +To manually register the `dataflow-tasklauncher` for either the `rabbit` or `kafka` binder, use one of the following URLs: -== Input - -a Task Launch Request including: - -* the task name (required and created as a task with the target Data Flow Server) -* deployment properties (key value pairs, optional). -* program arguments for the task (a list, optional). - -The message payload can be a JSON document: - -[source,json] ----- -{ - "name":"foo", - "deploymentProps": {"key1":"val1","key2":"val2"}, - "args":["--debug", "--foo", "bar"] -} ----- - -Minimally, it must include the task name. - -[source,json] ----- -{"name":"foo"} ----- - -=== Options - -The **$$tasklauncher-dataflow$$** $$sink$$ supports the following configuration properties: - -//tag::configuration-properties[] -Properties grouped by prefix: - - -=== spring.cloud.dataflow.client.authentication - -$$access-token$$:: $$OAuth2 Access Token.$$ *($$String$$, default: `$$$$`)* -$$client-id$$:: $$OAuth2 Client Id.$$ *($$String$$, default: `$$$$`)* -$$client-secret$$:: $$OAuth2 Client Secret.$$ *($$String$$, default: `$$$$`)* -$$scope$$:: $$OAuth2 Scopes.$$ *($$Set$$, default: `$$$$`)* -$$token-uri$$:: $$OAuth2 Token Uri.$$ *($$String$$, default: `$$$$`)* - -=== spring.cloud.dataflow.client.authentication.basic - -$$password$$:: $$The login password.$$ *($$String$$, default: `$$$$`)* -$$username$$:: $$The login username.$$ *($$String$$, default: `$$$$`)* - -=== spring.cloud.dataflow.client.authentication.oauth2 - -$$client-registration-id$$:: $$$$ *($$String$$, default: `$$$$`)* -$$password$$:: $$$$ *($$String$$, default: `$$$$`)* -$$username$$:: $$$$ *($$String$$, default: `$$$$`)* - -=== spring.cloud.dataflow.client - -$$enable-dsl$$:: $$Enable Data Flow DSL access.$$ *($$Boolean$$, default: `$$false$$`)* -$$server-uri$$:: $$The Data Flow server URI.$$ *($$String$$, default: `$$http://localhost:9393$$`)* -$$skip-ssl-validation$$:: $$Skip Ssl validation.$$ *($$Boolean$$, default: `$$false$$`)* - -=== tasklauncher - -$$platform-name$$:: $$The Spring Cloud Data Flow platform to use for launching tasks.$$ *($$String$$, default: `$$default$$`)* - -=== trigger - -$$initial-delay$$:: $$The initial delay in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)* -$$max-period$$:: $$The maximum polling period in milliseconds. Will be set to period if period > maxPeriod.$$ *($$Integer$$, default: `$$30000$$`)* -$$period$$:: $$The polling period in milliseconds.$$ *($$Integer$$, default: `$$1000$$`)* -//end::configuration-properties[] - -== Using the TaskLauncher -The tasklauncher sink consumes `LaunchRequest` messages, as described above, and launches a task using the target Data Flow server (given by `--spring.cloud.dataflow.client.server-uri`). -The task launcher periodically polls its input source for launch requests but will pause polling when the platform has reached it's concurrent task execution limit, given by `spring.cloud.dataflow.task.platform..accounts[].maximum-concurrent-tasks`. -This prevents the SCDF deployer's deployment platform from exhausting its resources under heavy task load. -The poller is scheduled using a `DynamicPeriodicTrigger`. By default the initial polling rate is 1 second, but may be configured to any duration. When polling is paused, or if there are no launch requests present, the trigger period will increase, applying exponential backoff, up to a configured maximum (30 seconds by default). - -NOTE: This version of the Data Flow task launcher requires SCDF version 2.4.x or higher - -The SCDF server may be configured to launch tasks on multiple platforms. -Each task launcher instance is configured for a single platform, given by the `platformName` property (`default` if not specified). -This limitation is enforced because if the server has multiple task platforms configured, it may be the case that some of its task platforms are at the limit and some are not. -In this situation, we can only consume the next launch request if we know for which task platform it is targeted. -For this reason, if the SCDF server is configured for multiple task platforms (or a single non-default platform), we assume that all launch requests are targeted for that platform. -The task launcher will set the required deployment property `spring.cloud.dataflow.task.platformName` if the request does not provide it. - -NOTE: If the request includes the deployment property `spring.cloud.dataflow.task.platformName`, and the value is not the same as the tasklauncher's `platformName`, the task launcher will throw an exception. - -To launch tasks on multiple platforms, you must configure a task launcher instance per platform and use a link:../router-sink/[router sink], or https://docs.spring.io/spring-cloud-stream/docs/current/reference/htmlsingle/#partitioning[partitioning strategy], to route requests to the correct instance. - -NOTE: When the poller is paused it puts pressure -on the message broker so some tuning will be necessary in extreme cases to balance resource utilization. - -=== Client Authentication - -If the Data Flow server requires authentication, the client must pass credentials with authorization to launch a task. -The Data Flow client supports both basic and OAuth2 authentication. - -For basic authentication set the username and password: - -``` ---spring.cloud.dataflow.client.authentication.basic.username= --spring.cloud.dataflow.client.authentication.basic.password= -``` - -For OAuth2 authentication, set the `client-id`, `client-secret`, and `token-uri` at a minimum. These values correspond to values set in the SCDF server's OAuth2 configuration. -For more details, see https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-security[the Security section in the Data Flow reference]. - -``` ---spring.cloud.dataflow.client.authentication.client-id= --spring.cloud.dataflow.client.authentication.client-secret= spring.cloud.dataflow.client.authentication.token-uri: -``` - -//end::ref-doc[] +maven://org.springframework.cloud:spring-cloud-dataflow-tasklauncher-sink-: +docker:springcloud/spring-cloud-dataflow-tasklauncher-sink-: diff --git a/applications/sink/tasklauncher-sink/pom.xml b/applications/sink/tasklauncher-sink/pom.xml deleted file mode 100644 index eed8ab85..00000000 --- a/applications/sink/tasklauncher-sink/pom.xml +++ /dev/null @@ -1,101 +0,0 @@ - - - 4.0.0 - tasklauncher-sink - 3.1.0-SNAPSHOT - tasklauncher-sink - tasklauncher sink app - jar - - - org.springframework.cloud.stream.app - stream-applications-core - 3.1.0-SNAPSHOT - - - - - - org.springframework.cloud.fn - tasklauncher-function - - - org.springframework.boot - spring-boot-configuration-processor - - - org.springframework.boot - spring-boot-starter-test - test - - - - - - - org.springframework.cloud - spring-cloud-dataflow-apps-docs-plugin - - - org.springframework.cloud - spring-cloud-dataflow-apps-generator-plugin - - - tasklauncher - sink - ${project.version} - org.springframework.cloud.stream.app.tasklauncher.sink.TaskLauncherSinkConfiguration.class - - - - - org.springframework.cloud.stream.app - tasklauncher-sink - ${project.version} - - - - - - - - - - - - - true - - spring-snapshots - Spring Snapshots - https://repo.spring.io/snapshot - - - - false - - spring-milestone-release - Spring Milestone Release - https://repo.spring.io/milestone - - - - - - true - - spring-snapshots - Spring Snapshots - https://repo.spring.io/snapshot - - - - false - - spring-milestones - Spring Milestones - https://repo.spring.io/milestone - - - diff --git a/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/LaunchRequestConsumer.java b/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/LaunchRequestConsumer.java deleted file mode 100644 index 9c17c964..00000000 --- a/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/LaunchRequestConsumer.java +++ /dev/null @@ -1,210 +0,0 @@ -/* - * Copyright 2020-2020 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.cloud.stream.app.tasklauncher.sink; - -import java.time.Duration; -import java.util.concurrent.ScheduledFuture; -import java.util.concurrent.atomic.AtomicBoolean; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.cloud.fn.tasklauncher.LaunchRequest; -import org.springframework.cloud.fn.tasklauncher.TaskLauncherFunction; -import org.springframework.cloud.stream.binder.PollableMessageSource; -import org.springframework.context.SmartLifecycle; -import org.springframework.core.ParameterizedTypeReference; -import org.springframework.integration.util.DynamicPeriodicTrigger; -import org.springframework.scheduling.concurrent.ConcurrentTaskScheduler; -import org.springframework.util.Assert; - -/** - * - * A Message consumer that submits received task {@link LaunchRequest}s to a Data Flow - * server. This polls a {@link PollableMessageSource} only if the Data Flow server is not - * at its concurrent task execution limit. - * - * The consumer runs as a {@link ScheduledFuture} , configured with a - * {@link DynamicPeriodicTrigger} to support exponential backoff up to a maximum period. - * Every period cycle, the poller first makes a REST call to the Data Flow server to check - * if it can accept a new task LaunchRequest before checking the Message source. The - * polling period will back off (increase) when either the server is not accepting - * requests or no request is received. - * - * The period will revert to its initial value whenever both a request is received and the - * DataFlow Server is accepting launch requests. The period remain at the maximum value - * when there are no requests to avoid hammering the Data Flow server for no reason. - * - * @author David Turanski - **/ -public class LaunchRequestConsumer implements SmartLifecycle { - private static final Log log = LogFactory.getLog(LaunchRequestConsumer.class); - - private static final int BACKOFF_MULTIPLE = 2; - - static final String TASK_PLATFORM_NAME = "spring.cloud.dataflow.task.platformName"; - - private final PollableMessageSource input; - - private final AtomicBoolean running = new AtomicBoolean(); - - private final AtomicBoolean paused = new AtomicBoolean(); - - private final DynamicPeriodicTrigger trigger; - - private final ConcurrentTaskScheduler taskScheduler; - - private final long initialPeriod; - - private final long maxPeriod; - - private volatile boolean autoStart = true; - - private final TaskLauncherFunction taskLauncherFunction; - - private ScheduledFuture scheduledFuture; - - public LaunchRequestConsumer(PollableMessageSource input, DynamicPeriodicTrigger trigger, - long maxPeriod, TaskLauncherFunction taskLauncherFunction) { - Assert.notNull(input, "`input` cannot be null."); - Assert.notNull(taskLauncherFunction, "`taskLauncherFunction` cannot be null."); - this.taskLauncherFunction = taskLauncherFunction; - this.input = input; - this.trigger = trigger; - this.initialPeriod = trigger.getDuration().toMillis(); - this.maxPeriod = maxPeriod; - this.taskScheduler = new ConcurrentTaskScheduler(); - } - - /* - * Polling loop - */ - ScheduledFuture consume() { - - return taskScheduler.schedule(() -> { - if (!isRunning()) { - return; - } - - if (taskLauncherFunction.platformIsAcceptingNewTasks()) { - if (paused.compareAndSet(true, false)) { - log.info("Polling resumed"); - } - - if (!input.poll(message -> { - LaunchRequest request = (LaunchRequest) message.getPayload(); - log.debug("Received a Task launch request - task name: " + request.getTaskName()); - taskLauncherFunction.apply(request); - }, new ParameterizedTypeReference() { - })) { - backoff("No task launch request received"); - } - else { - if (trigger.getDuration().toMillis() > initialPeriod) { - trigger.setDuration(Duration.ofMillis(initialPeriod)); - log.info(String.format("Polling period reset to %d ms.", trigger.getDuration().toMillis())); - } - } - } - else { - paused.set(true); - backoff("Polling paused"); - - } - }, trigger); - } - - @Override - public boolean isAutoStartup() { - return autoStart; - } - - public void setAutoStartup(boolean autoStart) { - this.autoStart = autoStart; - } - - @Override - public synchronized void stop(Runnable callback) { - if (callback != null) { - callback.run(); - } - this.stop(); - } - - @Override - public void start() { - if (running.compareAndSet(false, true)) { - this.scheduledFuture = consume(); - } - } - - @Override - public void stop() { - if (running.getAndSet(false)) { - this.scheduledFuture.cancel(false); - } - } - - @Override - public boolean isRunning() { - return running.get(); - } - - public boolean isPaused() { - return paused.get(); - } - - @Override - public int getPhase() { - return Integer.MAX_VALUE; - } - - private void backoff(String message) { - synchronized (trigger) { - if (trigger.getDuration().compareTo(Duration.ZERO) > 0 - && trigger.getDuration().compareTo(Duration.ofMillis(maxPeriod)) < 0) { - - Duration duration = trigger.getDuration(); - - if (duration.multipliedBy(BACKOFF_MULTIPLE).compareTo(Duration.ofMillis(maxPeriod)) <= 0) { - // If d >= 1, round to 1 seconds. - if (duration.getSeconds() == 1) { - duration = Duration.ofSeconds(1); - } - duration = duration.multipliedBy(BACKOFF_MULTIPLE); - } - else { - duration = Duration.ofMillis(maxPeriod); - } - if (trigger.getDuration().toMillis() < 1000) { - log.info(String.format(message + " - increasing polling period to %d ms.", duration.toMillis())); - } - else { - log.info( - String.format(message + "- increasing polling period to %d seconds.", - duration.getSeconds())); - } - - trigger.setDuration(duration); - } - else if (trigger.getDuration() == Duration.ofMillis(maxPeriod)) { - log.info(message); - } - } - } - -} diff --git a/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/PollingSink.java b/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/PollingSink.java deleted file mode 100644 index 36e9713d..00000000 --- a/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/PollingSink.java +++ /dev/null @@ -1,33 +0,0 @@ -/* - * Copyright 2018-2020 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.cloud.stream.app.tasklauncher.sink; - -import org.springframework.cloud.stream.annotation.Input; -import org.springframework.cloud.stream.binder.PollableMessageSource; - -/** - * @author David Turanski - **/ -public interface PollingSink { - /** - * The input name. - */ - String INPUT = "input"; - - @Input(PollingSink.INPUT) - PollableMessageSource input(); -} diff --git a/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkConfiguration.java b/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkConfiguration.java deleted file mode 100644 index 7662d90c..00000000 --- a/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkConfiguration.java +++ /dev/null @@ -1,99 +0,0 @@ -/* - * Copyright 2018-2020 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.cloud.stream.app.tasklauncher.sink; - -import java.time.Duration; - -import org.springframework.beans.BeansException; -import org.springframework.beans.factory.annotation.Value; -import org.springframework.beans.factory.config.BeanPostProcessor; -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.fn.tasklauncher.TaskLauncherFunction; -import org.springframework.cloud.fn.tasklauncher.TaskLauncherFunctionConfiguration; -import org.springframework.cloud.stream.annotation.EnableBinding; -import org.springframework.cloud.stream.binder.DefaultPollableMessageSource; -import org.springframework.cloud.stream.binder.PollableMessageSource; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Import; -import org.springframework.integration.util.DynamicPeriodicTrigger; -import org.springframework.messaging.Message; -import org.springframework.messaging.MessageChannel; -import org.springframework.messaging.MessageHeaders; -import org.springframework.messaging.support.ChannelInterceptor; -import org.springframework.messaging.support.MessageBuilder; - -/** - * Configuration class for the TaskLauncher Data Flow Sink. - * - * @author David Turanski - * @author Gunnar Hillert - */ -@EnableBinding(PollingSink.class) -@EnableConfigurationProperties({ TriggerProperties.class }) -@Import(TaskLauncherFunctionConfiguration.class) -public class TaskLauncherSinkConfiguration { - - @Value("${autostart:true}") - private boolean autoStart; - - @Bean - public DynamicPeriodicTrigger periodicTrigger(TriggerProperties triggerProperties) { - DynamicPeriodicTrigger trigger = new DynamicPeriodicTrigger(triggerProperties.getPeriod()); - trigger.setInitialDuration(Duration.ofMillis(triggerProperties.getInitialDelay())); - return trigger; - } - - /* - * For backward compatibility with spring-cloud-stream-2.1.x - */ - @Bean - public BeanPostProcessor addInterceptorToPollableMessageSource() { - return new BeanPostProcessor() { - @Override - public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException { - if (bean instanceof DefaultPollableMessageSource) { - DefaultPollableMessageSource pollableMessageSource = (DefaultPollableMessageSource) bean; - pollableMessageSource.addInterceptor(new ChannelInterceptor() { - @Override - public Message preSend(Message message, MessageChannel channel) { - Message newMessage = message; - if (message.getHeaders().containsKey("originalContentType")) { - newMessage = MessageBuilder.fromMessage(message) - .setHeader(MessageHeaders.CONTENT_TYPE, - message.getHeaders().get("originalContentType")) - .build(); - } - return newMessage; - } - }); - } - return bean; - } - }; - } - - @Bean - public LaunchRequestConsumer launchRequestConsumer(PollableMessageSource input, - TaskLauncherFunction taskLauncherFunction, DynamicPeriodicTrigger trigger, - TriggerProperties triggerProperties) { - - LaunchRequestConsumer consumer = new LaunchRequestConsumer(input, - trigger, triggerProperties.getMaxPeriod(), taskLauncherFunction); - consumer.setAutoStartup(autoStart); - return consumer; - } -} diff --git a/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TriggerProperties.java b/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TriggerProperties.java deleted file mode 100644 index 183334f4..00000000 --- a/applications/sink/tasklauncher-sink/src/main/java/org/springframework/cloud/stream/app/tasklauncher/sink/TriggerProperties.java +++ /dev/null @@ -1,78 +0,0 @@ -/* - * Copyright 2018-2020 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.cloud.stream.app.tasklauncher.sink; - -import javax.annotation.PostConstruct; -import javax.validation.constraints.Min; - -import org.springframework.boot.context.properties.ConfigurationProperties; -import org.springframework.validation.annotation.Validated; - -/** - * @author David Turanski - **/ -@ConfigurationProperties(prefix = "trigger") -@Validated -public class TriggerProperties { - /** - * The initial delay in milliseconds. - */ - private int initialDelay = 1000; - - /** - * The polling period in milliseconds. - */ - private int period = 1000; - - /** - * The maximum polling period in milliseconds. Will be set to period if period > - * maxPeriod. - */ - private int maxPeriod = 30000; - - @Min(0) - public int getInitialDelay() { - return initialDelay; - } - - public void setInitialDelay(int initialDelay) { - this.initialDelay = initialDelay; - } - - @Min(0) - public int getPeriod() { - return period; - } - - public void setPeriod(int period) { - this.period = period; - } - - @Min(1000) - public int getMaxPeriod() { - return maxPeriod; - } - - public void setMaxPeriod(int maxPeriod) { - this.maxPeriod = maxPeriod; - } - - @PostConstruct - public void checkMaxPeriod() { - maxPeriod = Integer.max(maxPeriod, period); - } -} diff --git a/applications/sink/tasklauncher-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties b/applications/sink/tasklauncher-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties deleted file mode 100644 index f6df9375..00000000 --- a/applications/sink/tasklauncher-sink/src/main/resources/META-INF/dataflow-configuration-metadata-whitelist.properties +++ /dev/null @@ -1,9 +0,0 @@ -configuration-properties.classes=org.springframework.cloud.fn.tasklauncher.TaskLauncherFunctionProperties,\ - org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties,\ - org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties$Authentication,\ - org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties$Authentication$Basic,\ - org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties$Authentication$Oauth2,\ - org.springframework.cloud.stream.app.tasklauncher.sink.TriggerProperties - - - diff --git a/applications/sink/tasklauncher-sink/src/main/resources/META-INF/dataflow-configuration-metadata.properties b/applications/sink/tasklauncher-sink/src/main/resources/META-INF/dataflow-configuration-metadata.properties deleted file mode 100644 index f6df9375..00000000 --- a/applications/sink/tasklauncher-sink/src/main/resources/META-INF/dataflow-configuration-metadata.properties +++ /dev/null @@ -1,9 +0,0 @@ -configuration-properties.classes=org.springframework.cloud.fn.tasklauncher.TaskLauncherFunctionProperties,\ - org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties,\ - org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties$Authentication,\ - org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties$Authentication$Basic,\ - org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties$Authentication$Oauth2,\ - org.springframework.cloud.stream.app.tasklauncher.sink.TriggerProperties - - - diff --git a/applications/sink/tasklauncher-sink/src/test/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkTests.java b/applications/sink/tasklauncher-sink/src/test/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkTests.java deleted file mode 100644 index 1aa1770d..00000000 --- a/applications/sink/tasklauncher-sink/src/test/java/org/springframework/cloud/stream/app/tasklauncher/sink/TaskLauncherSinkTests.java +++ /dev/null @@ -1,312 +0,0 @@ -/* - * Copyright 2020-2020 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.cloud.stream.app.tasklauncher.sink; - -import java.time.Duration; -import java.util.ArrayList; -import java.util.Collections; -import java.util.List; -import java.util.concurrent.CountDownLatch; -import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; -import java.util.function.Predicate; - -import com.fasterxml.jackson.core.JsonProcessingException; -import com.fasterxml.jackson.databind.ObjectMapper; -import org.junit.jupiter.api.BeforeEach; -import org.junit.jupiter.api.Test; -import org.mockito.stubbing.Answer; - -import org.springframework.beans.factory.BeanCreationException; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.test.context.runner.ApplicationContextRunner; -import org.springframework.cloud.dataflow.rest.client.DataFlowOperations; -import org.springframework.cloud.dataflow.rest.client.TaskOperations; -import org.springframework.cloud.dataflow.rest.resource.CurrentTaskExecutionsResource; -import org.springframework.cloud.dataflow.rest.resource.LauncherResource; -import org.springframework.cloud.fn.tasklauncher.LaunchRequest; -import org.springframework.cloud.stream.binder.test.TestChannelBinderConfiguration; -import org.springframework.context.ApplicationContext; -import org.springframework.context.annotation.Bean; -import org.springframework.core.env.Environment; -import org.springframework.hateoas.PagedModel; -import org.springframework.integration.IntegrationMessageHeaderAccessor; -import org.springframework.integration.acks.AcknowledgmentCallback; -import org.springframework.integration.core.MessageSource; -import org.springframework.integration.util.DynamicPeriodicTrigger; -import org.springframework.messaging.Message; -import org.springframework.messaging.SubscribableChannel; -import org.springframework.messaging.support.ErrorMessage; -import org.springframework.messaging.support.MessageBuilder; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.mockito.ArgumentMatchers.anyList; -import static org.mockito.ArgumentMatchers.anyMap; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.isNull; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.when; - -/** - * @author David Turanski - **/ -public class TaskLauncherSinkTests { - - private ApplicationContextRunner contextRunner; - - @BeforeEach - public void setUp() { - contextRunner = TestChannelBinderConfiguration.applicationContextRunner() - .withUserConfiguration(TaskLauncherSinkConfiguration.class, TestConfig.class); - } - - @Test - public void consumerPausesWhenMaxTaskExecutionsReached() { - contextRunner = contextRunner.withPropertyValues( - "trigger.period=10", - "trigger.initial-delay=0", - "autostart=false") - .run(context -> { - CurrentTaskExecutionsResource currentTaskExecutionsResource = currentTaskExecutionsResource( - context); - - LaunchRequestConsumer consumer = consumer(context); - CountDownLatch countDownLatch = countDownLatch(context); - DynamicPeriodicTrigger trigger = trigger(context); - - consumer.start(); - - assertThat(countDownLatch.await(1, TimeUnit.SECONDS)).isTrue(); - assertThat(currentTaskExecutionsResource.getRunningExecutionCount()).isEqualTo( - currentTaskExecutionsResource.getMaximumTaskExecutions()); - assertThat(eventually(c -> c.isPaused() && c.isRunning(), consumer)).isTrue(); - - currentTaskExecutionsResource.setRunningExecutionCount(0); - assertThat(eventually(c -> !c.isPaused(), consumer)).isTrue(); - }); - } - - @Test - public void exponentialBackOff() { - contextRunner.withPropertyValues("trigger.period=10", "trigger.initial-delay=0") - .run(context -> { - LaunchRequestConsumer consumer = consumer(context); - CurrentTaskExecutionsResource currentTaskExecutionsResource = currentTaskExecutionsResource( - context); - currentTaskExecutionsResource.setRunningExecutionCount( - currentTaskExecutionsResource.getMaximumTaskExecutions()); - DynamicPeriodicTrigger trigger = trigger(context); - - long waitTime = 0; - while (trigger.getDuration().compareTo(Duration.ofMillis(80)) < 0) { - Thread.sleep(10); - waitTime += 10; - assertThat(waitTime).isLessThan(1000); - } - assertThat(consumer.isPaused() && consumer.isRunning()).isTrue(); - }); - } - - @Test - public void backoffWhenNoMessages() { - - contextRunner.withPropertyValues( - "trigger.period=10", - "trigger.initial-delay=0", - "messageSourceDisabled=true", - "countDown=3") - .run(context -> { - CountDownLatch countDownLatch = countDownLatch(context); - CurrentTaskExecutionsResource currentTaskExecutionsResource = currentTaskExecutionsResource( - context); - DynamicPeriodicTrigger trigger = trigger(context); - - assertThat(countDownLatch.await(1, TimeUnit.SECONDS)).isTrue(); - assertThat(currentTaskExecutionsResource.getRunningExecutionCount()).isZero(); - assertThat(trigger.getDuration()).isGreaterThanOrEqualTo(Duration.ofMillis(40)); - }); - } - - @Test - public void launchRequestHasWrongPlatform() { - final AtomicBoolean passed = new AtomicBoolean(); - contextRunner.withPropertyValues( - "trigger.period=10", - "trigger.initial-delay=0", - "autostart=false", - "spring.cloud.stream.bindings.input.consumer.max-attempts=1", - "requestWrongPlatform=true") - .run(context -> { - - SubscribableChannel errorChannel = context.getBean("errorChannel", SubscribableChannel.class); - errorChannel.subscribe(message -> { - try { - assertThat(message).isInstanceOf(ErrorMessage.class); - ErrorMessage errorMessage = (ErrorMessage) message; - assertThat(errorMessage.getPayload()).isInstanceOf(Exception.class); - Exception exception = (Exception) message.getPayload(); - assertThat(exception.getCause().getMessage()).isEqualTo( - "Task Launch request for Task foo contains deployment property 'spring.cloud.dataflow" - + ".task.platformName=other' which does not match the platform configured for the Task" - + " Launcher: 'default'"); - passed.set(true); - } - catch (Exception e) { - } - }); - LaunchRequestConsumer consumer = consumer(context); - consumer.start(); - assertThat(eventually(c -> passed.get(), consumer)).isTrue(); - }); - } - - private CurrentTaskExecutionsResource currentTaskExecutionsResource(ApplicationContext context) { - CurrentTaskExecutionsResource currentTaskExecutionsResource = context - .getBean(CurrentTaskExecutionsResource.class); - currentTaskExecutionsResource.setRunningExecutionCount(0); - currentTaskExecutionsResource.setMaximumTaskExecutions(10); - return currentTaskExecutionsResource; - } - - private CountDownLatch countDownLatch(ApplicationContext context) { - return context.getBean(CountDownLatch.class); - } - - private LaunchRequestConsumer consumer(ApplicationContext context) { - return context.getBean(LaunchRequestConsumer.class); - } - - private DynamicPeriodicTrigger trigger(ApplicationContext context) { - return context.getBean(DynamicPeriodicTrigger.class); - } - - private synchronized boolean eventually(Predicate condition, - LaunchRequestConsumer consumer) { - final long MAX_WAIT = 1000; - long waitTime = 0; - long sleepTime = 10; - while (waitTime < MAX_WAIT) { - if (condition.test(consumer)) { - return true; - } - waitTime += sleepTime; - try { - Thread.sleep(sleepTime); - } - catch (InterruptedException e) { - Thread.interrupted(); - } - } - return condition.test(consumer); - } - - @SpringBootApplication - static class TestConfig { - - private TaskOperations taskOperations; - - private CurrentTaskExecutionsResource currentTaskExecutionsResource = new CurrentTaskExecutionsResource(); - - @Bean - public CurrentTaskExecutionsResource currentTaskExecutionsResource(Environment environment) { - currentTaskExecutionsResource.setMaximumTaskExecutions( - Integer.valueOf(environment.getProperty("maxExecutions", "10"))); - currentTaskExecutionsResource.setName("default"); - return currentTaskExecutionsResource; - } - - @Bean - public CountDownLatch countDownLatch(CurrentTaskExecutionsResource resource, Environment environment) { - return new CountDownLatch( - environment.containsProperty("countDown") ? Integer.valueOf(environment.getProperty("countDown")) - : resource.getMaximumTaskExecutions()); - } - - @Bean - DataFlowOperations dataFlowOperations(CurrentTaskExecutionsResource currentTaskExecutionsResource, - CountDownLatch latch) { - - DataFlowOperations dataFlowOperations; - taskOperations = mock(TaskOperations.class); - when(taskOperations.launch(anyString(), anyMap(), anyList(), isNull())) - .thenAnswer((Answer) invocation -> { - currentTaskExecutionsResource.setRunningExecutionCount( - currentTaskExecutionsResource.getRunningExecutionCount() + 1); - latch.countDown(); - return Long.valueOf(currentTaskExecutionsResource.getRunningExecutionCount()); - }); - - List launcherResources = new ArrayList<>(); - LauncherResource launcherResource0 = mock(LauncherResource.class); - when(launcherResource0.getName()).thenReturn("default"); - LauncherResource launcherResource1 = mock(LauncherResource.class); - when(launcherResource1.getName()).thenReturn("other"); - - when(taskOperations.currentTaskExecutions()).thenReturn( - Collections.singletonList(currentTaskExecutionsResource)); - LauncherResource launcherResource = mock(LauncherResource.class); - when(launcherResource.getName()).thenReturn("default"); - - launcherResources.add(launcherResource0); - launcherResources.add(launcherResource1); - - when(taskOperations.listPlatforms()) - .thenReturn(PagedModel.of(launcherResources, (PagedModel.PageMetadata) null)); - - dataFlowOperations = mock(DataFlowOperations.class); - when(dataFlowOperations.taskOperations()).thenReturn(taskOperations); - return dataFlowOperations; - } - - @Bean - public MessageSource testMessageSource(Environment environment, CountDownLatch countDownLatch, - ObjectMapper objectMapper) { - return () -> { - boolean messageSourceDisabled = Boolean.valueOf( - environment.getProperty("messageSourceDisabled", "false")); - LaunchRequest request = new LaunchRequest(); - request.setTaskName("foo"); - if (environment.getProperty("requestWrongPlatform", "false") - .equals("true")) { - request.getDeploymentProperties().put(LaunchRequestConsumer.TASK_PLATFORM_NAME, - "other"); - } - - Message message = null; - - if (messageSourceDisabled) { - countDownLatch.countDown(); - } - else { - try { - message = MessageBuilder.withPayload( - objectMapper.writeValueAsBytes(request)) - .setHeader("contentType", "application/json") - .setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, - (AcknowledgmentCallback) status -> { - }) - .build(); - } - catch (JsonProcessingException e) { - throw new BeanCreationException(e.getMessage(), e); - } - } - return message; - }; - } - } -} diff --git a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/KafkaTimeSourceTests.java b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/KafkaTimeSourceTests.java index 880baec3..dfacfa3f 100644 --- a/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/KafkaTimeSourceTests.java +++ b/applications/stream-applications-integration-tests/src/test/java/org/springframework/cloud/stream/app/integration/test/source/time/KafkaTimeSourceTests.java @@ -16,7 +16,6 @@ package org.springframework.cloud.stream.app.integration.test.source.time; -import org.junit.Ignore; import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.testcontainers.junit.jupiter.Container; diff --git a/functions/function-dependencies/pom.xml b/functions/function-dependencies/pom.xml index 6c4ca103..ffb63d70 100644 --- a/functions/function-dependencies/pom.xml +++ b/functions/function-dependencies/pom.xml @@ -266,11 +266,6 @@ task-launch-request-function ${revision} - - org.springframework.cloud.fn - tasklauncher-function - ${revision} - org.springframework.cloud.fn twitter-function diff --git a/functions/function/tasklauncher-function/README.adoc b/functions/function/tasklauncher-function/README.adoc index 019be1b4..c3d2685b 100644 --- a/functions/function/tasklauncher-function/README.adoc +++ b/functions/function/tasklauncher-function/README.adoc @@ -1,39 +1,5 @@ # TaskLauncher Function -This module provides a `Function` that uses the Data Flow REST client to launch a registered task on a configured https://docs.spring.io/spring-cloud-dataflow/docs/current/reference/htmlsingle/#configuration-local-tasks[task platform]. -The client must be configured to connect to a remote Data Flow Server, including any required authentication (see Configuration Options below). - -## Beans for injection - -You can import the `TaskLauncherFunctionConfiguration` configuration in a Spring Boot application and then inject the following bean. - -`taskLauncherFunction` - -You may inject this as `TaskLauncherFunction` which implements `Function>`. - -You can use `taskLauncherFunction` as a qualifier when injecting. - -Once injected, you can use the `apply` method of the `Function` to launch a task. -The function takes a link:src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java[LaunchRequest] as input. -This is a simple value object that specifies, at a minimum, the name of the task registered in Data Flow. -Optionally, you can pass command line arguments as a `List` and deployment properties as a `Map`. -The return value is an `Optional` containing the unique task ID of the launched instance if the launch request is successful. - -NOTE: This version of the tasklauncher requires Spring Cloud Data Flow version 2.4.x or higher - -## Configuration Options - -Specific properties, including the task platform name are prefixed with `tasklauncher`. - -For more information on the various options available, please see link:src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java[TaskLauncherFunctionProperties.java] - -Data Flow client configuration properties are prefixed with `spring.cloud.dataflow.client`. -Please see https://github.com/spring-cloud/spring-cloud-dataflow/blob/master/spring-cloud-dataflow-rest-client/src/main/java/org/springframework/cloud/dataflow/rest/client/config/DataFlowClientProperties.java[DataFlowClientProperties.java] for more details. - -## Examples - -See this link:src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java[test suite] for examples of how this function is used. - -## Other usage - -See this link:../../../applications/sink/tasklauncher-sink/README.adoc[README] where this function is used to create a Spring Cloud Stream application to submit task launch requests. \ No newline at end of file +[NOTE] +This module has been relocated to the https://github.com/spring-cloud/spring-cloud-dataflow/tree/main/spring-cloud-dataflow-tasklauncher/spring-cloud-dataflow-tasklauncher-function[spring-cloud-data-flow repository]. +As of SCDF 2.9.x, this function is included in the SCDF build to ensure version compatibility with the SCDF server going forward. diff --git a/functions/function/tasklauncher-function/pom.xml b/functions/function/tasklauncher-function/pom.xml deleted file mode 100644 index 71f886dc..00000000 --- a/functions/function/tasklauncher-function/pom.xml +++ /dev/null @@ -1,41 +0,0 @@ - - - 4.0.0 - - - org.springframework.cloud.fn - spring-functions-parent - ${revision} - ../../spring-functions-parent - - - tasklauncher-function - tasklauncher-function - Spring Native Function for applying filter SpEL expressions - - - 2.5.1.RELEASE - - - - - - org.springframework.cloud - spring-cloud-dataflow-rest-client - ${spring.cloud.dataflow.version} - - - org.springframework.cloud - spring-cloud-scheduler-spi - - - org.springframework.cloud - spring-cloud-skipper - - - - - - - diff --git a/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java b/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java deleted file mode 100644 index 54590e44..00000000 --- a/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/LaunchRequest.java +++ /dev/null @@ -1,66 +0,0 @@ -/* - * Copyright 2019-2020 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.cloud.fn.tasklauncher; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; - -import com.fasterxml.jackson.annotation.JsonProperty; - -import org.springframework.util.Assert; - -/** - * @author David Turanski - **/ - -public class LaunchRequest { - @JsonProperty("args") - private List commandlineArguments = new ArrayList<>(); - @JsonProperty("deploymentProps") - private Map deploymentProperties = new HashMap<>(); - @JsonProperty("name") - private String taskName; - - public List getCommandlineArguments() { - return commandlineArguments; - } - - public void setCommandlineArguments(List commandlineArguments) { - Assert.notNull(commandlineArguments, "'commandLineArguments' cannot be null."); - this.commandlineArguments = commandlineArguments; - } - - public Map getDeploymentProperties() { - return deploymentProperties; - } - - public void setDeploymentProperties(Map deploymentProperties) { - Assert.notNull(commandlineArguments, "'deploymentProperties' cannot be null."); - this.deploymentProperties = deploymentProperties; - } - - public String getTaskName() { - return taskName; - } - - public void setTaskName(String taskName) { - Assert.hasText(taskName, "'taskName' cannot be blank."); - this.taskName = taskName; - } -} diff --git a/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunction.java b/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunction.java deleted file mode 100644 index 097de153..00000000 --- a/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunction.java +++ /dev/null @@ -1,167 +0,0 @@ -/* - * Copyright 2019-2020 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.cloud.fn.tasklauncher; - -import java.util.ArrayList; -import java.util.HashMap; -import java.util.List; -import java.util.Map; -import java.util.Optional; -import java.util.function.Function; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.springframework.beans.factory.InitializingBean; -import org.springframework.cloud.dataflow.rest.client.TaskOperations; -import org.springframework.cloud.dataflow.rest.resource.CurrentTaskExecutionsResource; -import org.springframework.cloud.dataflow.rest.resource.LauncherResource; -import org.springframework.hateoas.PagedModel; -import org.springframework.util.Assert; -import org.springframework.util.StringUtils; - -/** - * - * A {@link Function} that submits a task {@link LaunchRequest} to a Data Flow server. - * This will check if the Data Flow task platform is at capacity. If not, will submit the - * task launch request, otherwise it will return and log a warning message. - * - * @author David Turanski - **/ -public class TaskLauncherFunction implements Function>, InitializingBean { - private static final Log log = LogFactory.getLog(TaskLauncherFunction.class); - - static final String TASK_PLATFORM_NAME = "spring.cloud.dataflow.task.platformName"; - - private final TaskOperations taskOperations; - - private String platformName = "default"; - - public TaskLauncherFunction(TaskOperations taskOperations) { - Assert.notNull(taskOperations, "`taskOperations` cannot be null."); - this.taskOperations = taskOperations; - } - - /** - * - * @param launchRequest the task launch request for the Data Flow server. - * @return an {@code Optional} containing the task Id if the request is accepted or - * empty otherwise. - */ - @Override - public Optional apply(LaunchRequest launchRequest) { - if (platformIsAcceptingNewTasks()) { - return Optional.of(launchTask(launchRequest)); - } - log.warn(String.format("Platform is at capacity. Did not submit task launch request for task %s.", - launchRequest.getTaskName())); - return Optional.empty(); - } - - public boolean platformIsAcceptingNewTasks() { - - boolean availableForNewTasks; - int maximumTaskExecutions = 0; - int runningExecutionCount = 0; - - List currentPlatforms = new ArrayList<>(); - - boolean validPlatform = false; - for (CurrentTaskExecutionsResource currentTaskExecutionsResource : taskOperations.currentTaskExecutions()) { - if (currentTaskExecutionsResource.getName().equals(platformName)) { - maximumTaskExecutions = currentTaskExecutionsResource.getMaximumTaskExecutions(); - runningExecutionCount = currentTaskExecutionsResource.getRunningExecutionCount(); - validPlatform = true; - } - currentPlatforms.add(currentTaskExecutionsResource.getName()); - } - - // Verify for each request as configuration may have changed on server. - assertValidPlatform(validPlatform, currentPlatforms); - - availableForNewTasks = runningExecutionCount < maximumTaskExecutions; - if (!availableForNewTasks) { - log.warn(String.format( - "The data Flow task platform %s has reached its concurrent task execution limit: (%d)", - platformName, - maximumTaskExecutions)); - } - - return availableForNewTasks; - - } - - private long launchTask(LaunchRequest request) { - String requestPlatformName = request.getDeploymentProperties().get(TASK_PLATFORM_NAME); - if (StringUtils.hasText(requestPlatformName) && !platformName.equals(requestPlatformName)) { - throw new IllegalStateException( - String.format( - "Task Launch request for Task %s contains deployment property '%s=%s' which does not " + - "match the platform configured for the Task Launcher: '%s'", - request.getTaskName(), - TASK_PLATFORM_NAME, - request.getDeploymentProperties().get(TASK_PLATFORM_NAME), - platformName)); - } - log.info(String.format("Launching Task %s on platform %s", request.getTaskName(), platformName)); - long taskId = taskOperations.launch(request.getTaskName(), - enrichDeploymentProperties(request.getDeploymentProperties()), - request.getCommandlineArguments(), null); - log.info(String.format("Launched Task %s - task ID is %d", request.getTaskName(), taskId)); - return taskId; - } - - private Map enrichDeploymentProperties(Map deploymentProperties) { - if (!deploymentProperties.containsKey(TASK_PLATFORM_NAME)) { - Map enrichedProperties = new HashMap<>(); - enrichedProperties.putAll(deploymentProperties); - enrichedProperties.put(TASK_PLATFORM_NAME, platformName); - return enrichedProperties; - } - return deploymentProperties; - } - - public void setPlatformName(String platformName) { - this.platformName = platformName; - } - - @Override - public void afterPropertiesSet() { - PagedModel launchers = taskOperations.listPlatforms(); - - boolean validPlatform = false; - List currentPlatforms = new ArrayList<>(); - - for (LauncherResource launcherResource : launchers) { - currentPlatforms.add(launcherResource.getName()); - if (launcherResource.getName().equals(platformName)) { - validPlatform = true; - } - } - - assertValidPlatform(validPlatform, currentPlatforms); - } - - private void assertValidPlatform(boolean validPlatform, List currentPlatforms) { - Assert.notEmpty(currentPlatforms, "The Data Flow Server has no task platforms configured"); - - Assert.isTrue(validPlatform, String.format( - "The task launcher's platform name '%s' does not match one of the Data Flow server's configured task " - + "platforms: [%s].", - platformName, StringUtils.collectionToCommaDelimitedString(currentPlatforms))); - } -} diff --git a/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionConfiguration.java b/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionConfiguration.java deleted file mode 100644 index 463eeeb3..00000000 --- a/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionConfiguration.java +++ /dev/null @@ -1,46 +0,0 @@ -/* - * Copyright 2019-2020 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.cloud.fn.tasklauncher; - -import org.springframework.boot.context.properties.EnableConfigurationProperties; -import org.springframework.cloud.dataflow.rest.client.DataFlowOperations; -import org.springframework.cloud.dataflow.rest.client.config.DataFlowClientProperties; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; - -/** - * Configuration class for the TaskLauncher Data Flow Sink. - * - * @author David Turanski - * @author Gunnar Hillert - */ -@Configuration -@EnableConfigurationProperties({TaskLauncherFunctionProperties.class, DataFlowClientProperties.class}) -public class TaskLauncherFunctionConfiguration { - - @Bean - public TaskLauncherFunction taskLauncherFunction( - DataFlowOperations dataFlowOperations, TaskLauncherFunctionProperties functionProperties) { - - if (dataFlowOperations.taskOperations() == null) { - throw new IllegalArgumentException("The SCDF server does not support task operations"); - } - TaskLauncherFunction function = new TaskLauncherFunction(dataFlowOperations.taskOperations()); - function.setPlatformName(functionProperties.getPlatformName()); - return function; - } -} diff --git a/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java b/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java deleted file mode 100644 index 632c64f1..00000000 --- a/functions/function/tasklauncher-function/src/main/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionProperties.java +++ /dev/null @@ -1,38 +0,0 @@ -/* - * Copyright 2019-2020 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.cloud.fn.tasklauncher; - -import org.springframework.boot.context.properties.ConfigurationProperties; - -/** - * @author David Turanski - **/ -@ConfigurationProperties("tasklauncher") -public class TaskLauncherFunctionProperties { - /** - * The Spring Cloud Data Flow platform to use for launching tasks. - */ - private String platformName = "default"; - - public String getPlatformName() { - return platformName; - } - - public void setPlatformName(String platformName) { - this.platformName = platformName; - } -} diff --git a/functions/function/tasklauncher-function/src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java b/functions/function/tasklauncher-function/src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java deleted file mode 100644 index 1fe763f8..00000000 --- a/functions/function/tasklauncher-function/src/test/java/org/springframework/cloud/fn/tasklauncher/TaskLauncherFunctionApplicationTests.java +++ /dev/null @@ -1,151 +0,0 @@ -/* - * Copyright 2019-2020 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.cloud.fn.tasklauncher; - -import java.util.Collections; -import java.util.Optional; - -import org.junit.jupiter.api.Test; - -import org.springframework.beans.factory.BeanCreationException; -import org.springframework.beans.factory.annotation.Autowired; -import org.springframework.boot.autoconfigure.SpringBootApplication; -import org.springframework.boot.test.context.SpringBootTest; -import org.springframework.boot.test.context.runner.ApplicationContextRunner; -import org.springframework.cloud.dataflow.rest.client.DataFlowOperations; -import org.springframework.cloud.dataflow.rest.client.TaskOperations; -import org.springframework.cloud.dataflow.rest.resource.CurrentTaskExecutionsResource; -import org.springframework.cloud.dataflow.rest.resource.LauncherResource; -import org.springframework.context.annotation.Bean; -import org.springframework.context.annotation.Configuration; -import org.springframework.context.annotation.Import; -import org.springframework.context.annotation.Profile; -import org.springframework.hateoas.PagedModel; - -import static org.assertj.core.api.Assertions.assertThat; -import static org.assertj.core.api.Assertions.assertThatExceptionOfType; -import static org.assertj.core.api.Assertions.assertThatIllegalStateException; -import static org.mockito.ArgumentMatchers.anyList; -import static org.mockito.ArgumentMatchers.anyMap; -import static org.mockito.ArgumentMatchers.anyString; -import static org.mockito.ArgumentMatchers.isNull; -import static org.mockito.Mockito.mock; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - -@SpringBootTest -public class TaskLauncherFunctionApplicationTests { - - @Autowired - private TaskLauncherFunction taskLauncherFunction; - - @Autowired - private TaskOperations taskOperations; - - @Test - public void successfulLaunch() { - LaunchRequest launchRequest = new LaunchRequest(); - launchRequest.setTaskName("someTask"); - setCurrentExecutionState(0); - Optional taskId = taskLauncherFunction.apply(launchRequest); - assertThat(taskId.isPresent()).isTrue(); - assertThat(taskId.get()).isEqualTo(1L); - - verify(taskOperations).launch("someTask", - Collections.singletonMap(TaskLauncherFunction.TASK_PLATFORM_NAME, "default"), - Collections.emptyList(), - null); - } - - @Test - public void taskPlatformAtCapacity() { - LaunchRequest launchRequest = new LaunchRequest(); - launchRequest.setTaskName("someTask"); - setCurrentExecutionState(3); - Optional taskId = taskLauncherFunction.apply(launchRequest); - assertThat(taskId.isPresent()).isFalse(); - } - - @Test - public void platformMismatch() { - LaunchRequest launchRequest = new LaunchRequest(); - launchRequest.setTaskName("someTask"); - launchRequest - .setDeploymentProperties(Collections.singletonMap(TaskLauncherFunction.TASK_PLATFORM_NAME, "other")); - setCurrentExecutionState(0); - assertThatIllegalStateException().isThrownBy(() -> taskLauncherFunction.apply(launchRequest)) - .withStackTraceContaining("does not match the platform configured for the Task Launcher"); - } - - private void setCurrentExecutionState(int runningExecutions) { - CurrentTaskExecutionsResource currentTaskExecutionsResource = new CurrentTaskExecutionsResource(); - currentTaskExecutionsResource.setMaximumTaskExecutions(3); - currentTaskExecutionsResource.setRunningExecutionCount(runningExecutions); - currentTaskExecutionsResource.setName("default"); - when(taskOperations.currentTaskExecutions()) - .thenReturn(Collections.singletonList(currentTaskExecutionsResource)); - when(taskOperations.launch(anyString(), anyMap(), anyList(), isNull())).thenReturn(1L); - } - - @Test - public void noLaunchersConfigured() { - ApplicationContextRunner contextRunner = new ApplicationContextRunner().withUserConfiguration(TestConfig.class); - assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> contextRunner - .withPropertyValues("spring.profiles.active=nolaunchers") - .run(context -> context.start())) - .withCauseInstanceOf(BeanCreationException.class) - .withRootCauseInstanceOf(IllegalArgumentException.class) - .withStackTraceContaining("The Data Flow Server has no task platforms configured"); - } - - @Configuration - @Import(TaskLauncherFunctionConfiguration.class) - static class TestConfig { - - @Bean - @Profile("default") - TaskOperations taskOperations() { - TaskOperations taskOperations = mock(TaskOperations.class); - LauncherResource launcherResource = mock(LauncherResource.class); - when(launcherResource.getName()).thenReturn("default"); - - when(taskOperations.listPlatforms()).thenReturn(PagedModel.of( - Collections.singletonList(launcherResource), (PagedModel.PageMetadata) null)); - return taskOperations; - } - - @Bean - @Profile("nolaunchers") - TaskOperations taskOperationsNoLaunchers() { - TaskOperations taskOperations = mock(TaskOperations.class); - when(taskOperations.listPlatforms()).thenReturn(PagedModel.of( - Collections.emptyList(), (PagedModel.PageMetadata) null)); - return taskOperations; - } - - @Bean - DataFlowOperations dataFlowOperations(TaskOperations taskOperations) { - DataFlowOperations dataFlowOperations = mock(DataFlowOperations.class); - when(dataFlowOperations.taskOperations()).thenReturn(taskOperations); - return dataFlowOperations; - } - } - - @SpringBootApplication - static class TaskLauncherFunctionTestApplication { - } -} diff --git a/functions/pom.xml b/functions/pom.xml index af72c93d..d951190b 100644 --- a/functions/pom.xml +++ b/functions/pom.xml @@ -64,7 +64,6 @@ function/spel-function function/payload-converter-function function/splitter-function - function/tasklauncher-function function/task-launch-request-function function/twitter-function function/image-recognition-function diff --git a/stream-applications-release-train/stream-applications-descriptor/pom.xml b/stream-applications-release-train/stream-applications-descriptor/pom.xml index 5b47032c..a77a15f8 100644 --- a/stream-applications-release-train/stream-applications-descriptor/pom.xml +++ b/stream-applications-release-train/stream-applications-descriptor/pom.xml @@ -125,8 +125,6 @@ "${s3-sink.version}".contains('SNAPSHOT') ? 'latest' : "${s3-sink.version}" pom.properties['sftp-sink-docker.tag']= "${sftp-sink.version}".contains('SNAPSHOT') ? 'latest' : "${sftp-sink.version}" - pom.properties['tasklauncher-sink-docker.tag']= - "${tasklauncher-sink.version}".contains('SNAPSHOT') ? 'latest' : "${tasklauncher-sink.version}" pom.properties['tcp-sink-docker.tag']= "${tcp-sink.version}".contains('SNAPSHOT') ? 'latest' : "${tcp-sink.version}" pom.properties['throughput-sink-docker.tag']= diff --git a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-docker.properties b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-docker.properties index 98f4f474..bb976687 100644 --- a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-docker.properties +++ b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-docker.properties @@ -38,7 +38,6 @@ sink.router=docker:springcloudstream/router-sink-kafka:@router-sink-docker.tag@ sink.rsocket=docker:springcloudstream/rsocket-sink-kafka:@rsocket-sink-docker.tag@ sink.s3=docker:springcloudstream/s3-sink-kafka:@s3-sink-docker.tag@ sink.sftp=docker:springcloudstream/sftp-sink-kafka:@sftp-sink-docker.tag@ -sink.tasklauncher=docker:springcloudstream/tasklauncher-sink-kafka:@tasklauncher-sink-docker.tag@ sink.tcp=docker:springcloudstream/tcp-sink-kafka:@tcp-sink-docker.tag@ sink.throughput=docker:springcloudstream/throughput-sink-kafka:@throughput-sink-docker.tag@ sink.twitter-message=docker:springcloudstream/twitter-message-sink-kafka:@twitter-message-sink-docker.tag@ diff --git a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-harbor.properties b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-harbor.properties index 7555f7bc..237bcd7e 100644 --- a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-harbor.properties +++ b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-harbor.properties @@ -37,7 +37,6 @@ sink.router=docker:projects.registry.vmware.com/springcloudstream/router-sink-ka sink.rsocket=docker:projects.registry.vmware.com/springcloudstream/rsocket-sink-kafka:@apps.harbor.version@ sink.s3=docker:projects.registry.vmware.com/springcloudstream/s3-sink-kafka:@apps.harbor.version@ sink.sftp=docker:projects.registry.vmware.com/springcloudstream/sftp-sink-kafka:@apps.harbor.version@ -sink.tasklauncher=docker:projects.registry.vmware.com/springcloudstream/tasklauncher-sink-kafka:@apps.harbor.version@ sink.tcp=docker:projects.registry.vmware.com/springcloudstream/tcp-sink-kafka:@apps.harbor.version@ sink.throughput=docker:projects.registry.vmware.com/springcloudstream/throughput-sink-kafka:@apps.harbor.version@ sink.twitter-message=docker:projects.registry.vmware.com/springcloudstream/twitter-message-sink-kafka:@apps.harbor.version@ diff --git a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-maven-repo-url.properties b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-maven-repo-url.properties index 4411f5af..d5bbc8ac 100644 --- a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-maven-repo-url.properties +++ b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-maven-repo-url.properties @@ -78,8 +78,6 @@ sink.s3=https://@repo-spring-io@/org/springframework/cloud/stream/app/s3-sink-ka sink.s3.metadata=https://@repo-spring-io@/org/springframework/cloud/stream/app/s3-sink-kafka/@s3-sink.version@/s3-sink-kafka-@s3-sink.version@-metadata.jar sink.sftp=https://@repo-spring-io@/org/springframework/cloud/stream/app/sftp-sink-kafka/@sftp-sink.version@/sftp-sink-kafka-@sftp-sink.version@.jar sink.sftp.metadata=https://@repo-spring-io@/org/springframework/cloud/stream/app/sftp-sink-kafka/@sftp-sink.version@/sftp-sink-kafka-@sftp-sink.version@-metadata.jar -sink.tasklauncher=https://@repo-spring-io@/org/springframework/cloud/stream/app/tasklauncher-sink-kafka/@tasklauncher-sink.version@/tasklauncher-sink-kafka-@tasklauncher-sink.version@.jar -sink.tasklauncher.metadata=https://@repo-spring-io@/org/springframework/cloud/stream/app/tasklauncher-sink-kafka/@tasklauncher-sink.version@/tasklauncher-sink-kafka-@tasklauncher-sink.version@-metadata.jar sink.tcp=https://@repo-spring-io@/org/springframework/cloud/stream/app/tcp-sink-kafka/@tcp-sink.version@/tcp-sink-kafka-@tcp-sink.version@.jar sink.tcp.metadata=https://@repo-spring-io@/org/springframework/cloud/stream/app/tcp-sink-kafka/@tcp-sink.version@/tcp-sink-kafka-@tcp-sink.version@-metadata.jar sink.throughput=https://@repo-spring-io@/org/springframework/cloud/stream/app/throughput-sink-kafka/@throughput-sink.version@/throughput-sink-kafka-@throughput-sink.version@.jar diff --git a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-maven.properties b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-maven.properties index 471f27ac..ebe0b2d2 100644 --- a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-maven.properties +++ b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/kafka-apps-maven.properties @@ -78,8 +78,6 @@ sink.s3=maven://org.springframework.cloud.stream.app:s3-sink-kafka:@s3-sink.vers sink.s3.metadata=maven://org.springframework.cloud.stream.app:s3-sink-kafka:jar:metadata:@s3-sink.version@ sink.sftp=maven://org.springframework.cloud.stream.app:sftp-sink-kafka:@sftp-sink.version@ sink.sftp.metadata=maven://org.springframework.cloud.stream.app:sftp-sink-kafka:jar:metadata:@sftp-sink.version@ -sink.tasklauncher=maven://org.springframework.cloud.stream.app:tasklauncher-sink-kafka:@tasklauncher-sink.version@ -sink.tasklauncher.metadata=maven://org.springframework.cloud.stream.app:tasklauncher-sink-kafka:jar:metadata:@tasklauncher-sink.version@ sink.tcp=maven://org.springframework.cloud.stream.app:tcp-sink-kafka:@tcp-sink.version@ sink.tcp.metadata=maven://org.springframework.cloud.stream.app:tcp-sink-kafka:jar:metadata:@tcp-sink.version@ sink.throughput=maven://org.springframework.cloud.stream.app:throughput-sink-kafka:@throughput-sink.version@ diff --git a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-docker.properties b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-docker.properties index 0fcaf0ec..81e8fc9c 100644 --- a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-docker.properties +++ b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-docker.properties @@ -38,7 +38,6 @@ sink.router=docker:springcloudstream/router-sink-rabbit:@router-sink-docker.tag@ sink.rsocket=docker:springcloudstream/rsocket-sink-rabbit:@rsocket-sink-docker.tag@ sink.s3=docker:springcloudstream/s3-sink-rabbit:@s3-sink-docker.tag@ sink.sftp=docker:springcloudstream/sftp-sink-rabbit:@sftp-sink-docker.tag@ -sink.tasklauncher=docker:springcloudstream/tasklauncher-sink-rabbit:@tasklauncher-sink-docker.tag@ sink.tcp=docker:springcloudstream/tcp-sink-rabbit:@tcp-sink-docker.tag@ sink.throughput=docker:springcloudstream/throughput-sink-rabbit:@throughput-sink-docker.tag@ sink.twitter-message=docker:springcloudstream/twitter-message-sink-rabbit:@twitter-message-sink-docker.tag@ diff --git a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-harbor.properties b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-harbor.properties index 7bbed5bb..fe2ace86 100644 --- a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-harbor.properties +++ b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-harbor.properties @@ -37,7 +37,6 @@ sink.router=docker:projects.registry.vmware.com/springcloudstream/router-sink-ra sink.rsocket=docker:projects.registry.vmware.com/springcloudstream/rsocket-sink-rabbit:@apps.harbor.version@ sink.s3=docker:projects.registry.vmware.com/springcloudstream/s3-sink-rabbit:@apps.harbor.version@ sink.sftp=docker:projects.registry.vmware.com/springcloudstream/sftp-sink-rabbit:@apps.harbor.version@ -sink.tasklauncher=docker:projects.registry.vmware.com/springcloudstream/tasklauncher-sink-rabbit:@apps.harbor.version@ sink.tcp=docker:projects.registry.vmware.com/springcloudstream/tcp-sink-rabbit:@apps.harbor.version@ sink.throughput=docker:projects.registry.vmware.com/springcloudstream/throughput-sink-rabbit:@apps.harbor.version@ sink.twitter-message=docker:projects.registry.vmware.com/springcloudstream/twitter-message-sink-rabbit:@apps.harbor.version@ diff --git a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-maven-repo-url.properties b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-maven-repo-url.properties index 9b4ba554..e41cc382 100644 --- a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-maven-repo-url.properties +++ b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-maven-repo-url.properties @@ -78,8 +78,6 @@ sink.s3=https://@repo-spring-io@/org/springframework/cloud/stream/app/s3-sink-ra sink.s3.metadata=https://@repo-spring-io@/org/springframework/cloud/stream/app/s3-sink-rabbit/@s3-sink.version@/s3-sink-rabbit-@s3-sink.version@-metadata.jar sink.sftp=https://@repo-spring-io@/org/springframework/cloud/stream/app/sftp-sink-rabbit/@sftp-sink.version@/sftp-sink-rabbit-@sftp-sink.version@.jar sink.sftp.metadata=https://@repo-spring-io@/org/springframework/cloud/stream/app/sftp-sink-rabbit/@sftp-sink.version@/sftp-sink-rabbit-@sftp-sink.version@-metadata.jar -sink.tasklauncher=https://@repo-spring-io@/org/springframework/cloud/stream/app/tasklauncher-sink-rabbit/@tasklauncher-sink.version@/tasklauncher-sink-rabbit-@tasklauncher-sink.version@.jar -sink.tasklauncher.metadata=https://@repo-spring-io@/org/springframework/cloud/stream/app/tasklauncher-sink-rabbit/@tasklauncher-sink.version@/tasklauncher-sink-rabbit-@tasklauncher-sink.version@-metadata.jar sink.tcp=https://@repo-spring-io@/org/springframework/cloud/stream/app/tcp-sink-rabbit/@tcp-sink.version@/tcp-sink-rabbit-@tcp-sink.version@.jar sink.tcp.metadata=https://@repo-spring-io@/org/springframework/cloud/stream/app/tcp-sink-rabbit/@tcp-sink.version@/tcp-sink-rabbit-@tcp-sink.version@-metadata.jar sink.throughput=https://@repo-spring-io@/org/springframework/cloud/stream/app/throughput-sink-rabbit/@throughput-sink.version@/throughput-sink-rabbit-@throughput-sink.version@.jar diff --git a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-maven.properties b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-maven.properties index e00d402e..0040abaf 100644 --- a/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-maven.properties +++ b/stream-applications-release-train/stream-applications-descriptor/src/main/resources/META-INF/rabbit-apps-maven.properties @@ -78,8 +78,6 @@ sink.s3=maven://org.springframework.cloud.stream.app:s3-sink-rabbit:@s3-sink.ver sink.s3.metadata=maven://org.springframework.cloud.stream.app:s3-sink-rabbit:jar:metadata:@s3-sink.version@ sink.sftp=maven://org.springframework.cloud.stream.app:sftp-sink-rabbit:@sftp-sink.version@ sink.sftp.metadata=maven://org.springframework.cloud.stream.app:sftp-sink-rabbit:jar:metadata:@sftp-sink.version@ -sink.tasklauncher=maven://org.springframework.cloud.stream.app:tasklauncher-sink-rabbit:@tasklauncher-sink.version@ -sink.tasklauncher.metadata=maven://org.springframework.cloud.stream.app:tasklauncher-sink-rabbit:jar:metadata:@tasklauncher-sink.version@ sink.tcp=maven://org.springframework.cloud.stream.app:tcp-sink-rabbit:@tcp-sink.version@ sink.tcp.metadata=maven://org.springframework.cloud.stream.app:tcp-sink-rabbit:jar:metadata:@tcp-sink.version@ sink.throughput=maven://org.springframework.cloud.stream.app:throughput-sink-rabbit:@throughput-sink.version@