You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@inlong.apache.org by al...@apache.org on 2024/06/22 07:00:56 UTC
(inlong) branch dev-offline-sync updated (2a08660a07 -> c1bda980b6)
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a change to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
discard 2a08660a07 [INLONG-10360][Manager] Combine schedule state transition with group operations (#10445)
discard c2f14111f7 [INLONG-10436][Manager] Move schedule configuration from stream to group (#10437)
discard 0eb6fa750d [INLONG-10396][Manager] Support build-in schedule base on quartz (#10412)
discard 4d5c873efb [INLONG-10395][Manager] Add interface for schedule client and engine (#10397)
discard 8bd3a20cb4 [INLONG-10247][Manager] Support schedule information management for offline sync (#10254)
discard 36afe37602 [INLONG-9862][Manager] Support submit flink job for offline sync (#9865)
discard fafcece220 [INLONG-9822][Manager] Support flink job runtime execution mode configuration (#9823)
discard 2674a5645c [INLONG-9813][Manager] Support offline data sync management (#9814)
discard 051c9716c6 [INLONG-9781][Manager] Add offline sync task type definition (#9787)
add 466cf626fc [INLONG-10472][Manager] Use audit SDK to obtain audit proxy URL (#10473)
add bb6cc76a02 [INLONG-10470][Audit] Optimize Audit Proxy configuration update retry logic (#10471)
add cc9466ff1e [INLONG-10449][Dashboard] The field template for selecting a new data (#10460)
add 30367cf465 [INLONG-10478][Audit] Default to enable audit for the Docker and Standalone deployment (#10479)
add 79eba735ed [INLONG-10476][Agent] The audit address should only be obtained from the manager (#10477)
add e7dd1d05ef [INLONG-10474][Manager] Restrict sortTaskName and sortConsumeName the same with datanodeName when migrate Inlong Group (#10475)
add fa0858c4c6 [INLONG-10482][Dashboard] Fix the front-end parameters on the approval page are incorrect (#10483)
add 4ee78b7260 [INLONG-10481][Audit] Optimize Audit domain management (#10486)
add cee09a6579 [INLONG-10487][Manager] Fix the error of not return success ListenerResult (#10488)
add b8da024cde [INLONG-10484][Manager] Refactor code in manager-service module (#10485)
new 565a67ba17 [INLONG-9781][Manager] Add offline sync task type definition (#9787)
new 0bae684745 [INLONG-9813][Manager] Support offline data sync management (#9814)
new b83a4831d4 [INLONG-9822][Manager] Support flink job runtime execution mode configuration (#9823)
new dc8aa2ef5e [INLONG-9862][Manager] Support submit flink job for offline sync (#9865)
new 2afe763e0d [INLONG-10247][Manager] Support schedule information management for offline sync (#10254)
new 1b2e6ef5ca [INLONG-10395][Manager] Add interface for schedule client and engine (#10397)
new a418ea3502 [INLONG-10396][Manager] Support build-in schedule base on quartz (#10412)
new 151c5f4ea2 [INLONG-10436][Manager] Move schedule configuration from stream to group (#10437)
new c1bda980b6 [INLONG-10360][Manager] Combine schedule state transition with group operations (#10445)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (2a08660a07)
\
N -- N -- N refs/heads/dev-offline-sync (c1bda980b6)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 9 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
bin/init-config.sh | 10 +-
bin/inlong-daemon | 2 +-
conf/inlong.conf | 2 -
docker/docker-compose/docker-compose.yml | 5 +-
.../inlong/agent/constant/AgentConstants.java | 3 -
.../inlong/agent/metrics/audit/AuditUtils.java | 33 +-
inlong-agent/agent-docker/Dockerfile | 3 -
inlong-agent/agent-docker/agent-docker.sh | 2 -
.../agent/plugin/store/RocksDBStoreImpl.java | 2 -
.../apache/inlong/audit/entity/AuditComponent.java | 2 +-
.../apache/inlong/audit/utils/ThreadUtils.java} | 20 +-
inlong-audit/audit-docker/Dockerfile | 3 +
inlong-audit/audit-docker/audit-docker.sh | 3 +
.../apache/inlong/audit/heartbeat/Heartbeat.java | 135 ---------
.../org/apache/inlong/audit/node/Application.java | 5 -
.../org/apache/inlong/audit/AuditReporterImpl.java | 4 +-
.../org/apache/inlong/audit/send/ProxyManager.java | 51 +++-
.../apache/inlong/audit/cache/AuditProxyCache.java | 236 ++++-----------
.../inlong/audit/config/OpenApiConstants.java | 7 -
.../inlong/audit/config/ProxyConstants.java} | 27 +-
.../apache/inlong/audit/config/SqlConstants.java | 65 +---
.../org/apache/inlong/audit/entities/ApiType.java | 2 +-
.../inlong/audit/heartbeat/ProxyHeartbeat.java | 96 ------
.../apache/inlong/audit/service/ApiService.java | 25 +-
.../org/apache/inlong/audit/sink/JdbcSink.java | 47 ++-
inlong-audit/conf/audit-service.properties | 8 +-
inlong-audit/sql/apache_inlong_audit_mysql.sql | 54 +---
.../plugins/groups/common/GroupDataTemplateInfo.ts | 1 +
.../plugins/streams/common/StreamDefaultInfo.ts | 34 +--
.../src/ui/components/EditableTable/index.tsx | 37 ++-
.../ui/components/SelectTemplateModal/index.tsx | 147 +++++++++
inlong-dashboard/src/ui/locales/cn.json | 4 +
inlong-dashboard/src/ui/locales/en.json | 4 +
.../src/ui/pages/GroupDataTemplate/CreateModal.tsx | 2 +-
.../GroupDetail/DataStream/StreamItemModal.tsx | 22 --
.../src/ui/pages/Process/Approvals/config.tsx | 3 +-
inlong-dataproxy/dataproxy-docker/Dockerfile | 3 -
.../dataproxy-docker/dataproxy-docker.sh | 3 +-
inlong-manager/manager-docker/Dockerfile | 1 -
inlong-manager/manager-docker/manager-docker.sh | 1 -
.../plugin/listener/StartupSortListener.java | 2 +-
inlong-manager/manager-pojo/pom.xml | 11 +
.../manager/pojo/audit/AuditProxyResponse.java | 12 +-
inlong-manager/manager-service/pom.xml | 11 -
.../service/cluster/InlongClusterService.java | 120 --------
.../service/cluster/InlongClusterServiceImpl.java | 331 ---------------------
.../node/AgentClusterNodeInstallOperator.java | 4 -
.../inlong/manager/service/core/AuditService.java | 2 +-
.../service/core/impl/AuditServiceImpl.java | 2 +-
.../resource/sort/SortFlinkConfigOperator.java | 10 +-
.../manager/service/sink/StreamSinkService.java | 37 ---
.../service/sink/StreamSinkServiceImpl.java | 174 +----------
.../service/source/StreamSourceService.java | 36 ---
.../service/source/StreamSourceServiceImpl.java | 130 +-------
.../service/stream/InlongStreamService.java | 41 ---
.../service/stream/InlongStreamServiceImpl.java | 151 ----------
.../service/tenant/InlongTenantServiceImpl.java | 3 +
.../service/transform/StreamTransformService.java | 38 ---
.../transform/StreamTransformServiceImpl.java | 122 --------
.../manager/web/controller/AuditController.java | 2 +-
.../web/controller/InlongStreamController.java | 2 +-
.../web/controller/StreamSinkController.java | 2 +-
.../web/controller/StreamSourceController.java | 2 +-
.../web/controller/openapi/AuditController.java | 2 +-
.../openapi/OpenInLongClusterController.java | 27 +-
.../openapi/OpenInLongStreamController.java | 8 +-
.../openapi/OpenStreamSinkController.java | 8 +-
.../openapi/OpenStreamSourceController.java | 8 +-
.../openapi/OpenStreamTransformController.java | 9 +-
.../src/main/resources/application-dev.properties | 3 -
.../src/main/resources/application-prod.properties | 3 -
.../src/main/resources/application-test.properties | 3 -
72 files changed, 470 insertions(+), 1960 deletions(-)
copy inlong-audit/{audit-service/src/main/java/org/apache/inlong/audit/selector/SelectorChangeListenerImpl.java => audit-common/src/main/java/org/apache/inlong/audit/utils/ThreadUtils.java} (70%)
delete mode 100644 inlong-audit/audit-proxy/src/main/java/org/apache/inlong/audit/heartbeat/Heartbeat.java
copy inlong-audit/{audit-store/src/main/java/org/apache/inlong/audit/config/StoreConfig.java => audit-service/src/main/java/org/apache/inlong/audit/config/ProxyConstants.java} (55%)
delete mode 100644 inlong-audit/audit-service/src/main/java/org/apache/inlong/audit/heartbeat/ProxyHeartbeat.java
create mode 100644 inlong-dashboard/src/ui/components/SelectTemplateModal/index.tsx
(inlong) 07/09: [INLONG-10396][Manager] Support build-in schedule base on quartz (#10412)
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit a418ea3502ea80daa0385751b9858ee426890a83
Author: AloysZhang <al...@apache.org>
AuthorDate: Thu Jun 13 22:44:42 2024 +0800
[INLONG-10396][Manager] Support build-in schedule base on quartz (#10412)
---
.../inlong/manager/dao/entity/ScheduleEntity.java | 2 +-
.../inlong/manager/pojo/schedule/ScheduleInfo.java | 4 +-
.../manager/pojo/schedule/ScheduleInfoRequest.java | 4 +-
inlong-manager/manager-schedule/pom.xml | 31 ++++
.../org/apache/inlong/schedule/ScheduleType.java | 45 ++++++
.../org/apache/inlong/schedule/ScheduleUnit.java | 48 ++++++
.../exception/QuartzScheduleException.java | 32 ++++
.../schedule/quartz/QuartzOfflineSyncJob.java | 40 +++++
.../schedule/quartz/QuartzScheduleClient.java | 50 +++++++
.../schedule/quartz/QuartzScheduleEngine.java | 156 ++++++++++++++++++++
.../schedule/quartz/QuartzSchedulerListener.java | 137 +++++++++++++++++
.../apache/inlong/schedule/util/ScheduleUtils.java | 139 +++++++++++++++++
.../apache/inlong/schedule/BaseScheduleTest.java | 116 +++++++++++++++
.../org/apache/inlong/schedule/quartz/MockJob.java | 48 ++++++
.../schedule/quartz/QuartzScheduleEngineTest.java | 161 ++++++++++++++++++++
.../inlong/schedule/util/ScheduleUtilsTest.java | 164 +++++++++++++++++++++
.../manager-schedule/src/test/resources/log4j2.xml | 46 ++++++
.../main/resources/h2/apache_inlong_manager.sql | 2 +-
.../manager-web/sql/apache_inlong_manager.sql | 2 +-
inlong-manager/manager-web/sql/changes-1.13.0.sql | 2 +-
inlong-manager/pom.xml | 2 +-
21 files changed, 1222 insertions(+), 9 deletions(-)
diff --git a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/ScheduleEntity.java b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/ScheduleEntity.java
index 75237343cb..b2d49cd43c 100644
--- a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/ScheduleEntity.java
+++ b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/ScheduleEntity.java
@@ -34,7 +34,7 @@ public class ScheduleEntity implements Serializable {
// schedule type, support [normal, crontab], 0 for normal and 1 for crontab
private Integer scheduleType;
// time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
- // M=month, W=week, D=day, H=hour, M=minute, O=oneway
+ // Y=year, M=month, W=week, D=day, H=hour, I=minute, O=oneway
private String scheduleUnit;
private Integer scheduleInterval;
// schedule start time, long type timestamp
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java
index 13afce70a4..2386d817bb 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java
@@ -50,7 +50,7 @@ public class ScheduleInfo {
private Integer scheduleType;
// time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
- // M=month, W=week, D=day, H=hour, M=minute, O=oneway
+ // Y=year, M=month, W=week, D=day, H=hour, I=minute, O=oneway
@ApiModelProperty("TimeUnit for schedule interval")
private String scheduleUnit;
@@ -73,7 +73,7 @@ public class ScheduleInfo {
private Integer taskParallelism;
@ApiModelProperty("Schedule task parallelism")
- private Integer crontabExpression;
+ private String crontabExpression;
@ApiModelProperty(value = "Version number")
@NotNull(groups = UpdateValidation.class, message = "version cannot be null")
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java
index eff1660719..b3c117da9a 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java
@@ -44,7 +44,7 @@ public class ScheduleInfoRequest {
private Integer scheduleType;
// time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
- // M=month, W=week, D=day, H=hour, M=minute, O=oneway
+ // Y=year, M=month, W=week, D=day, H=hour, I=minute, O=oneway
@ApiModelProperty("TimeUnit for schedule interval")
private String scheduleUnit;
@@ -67,7 +67,7 @@ public class ScheduleInfoRequest {
private Integer taskParallelism;
@ApiModelProperty("Schedule task parallelism")
- private Integer crontabExpression;
+ private String crontabExpression;
@ApiModelProperty(value = "Version number")
@NotNull(groups = UpdateValidation.class, message = "version cannot be null")
diff --git a/inlong-manager/manager-schedule/pom.xml b/inlong-manager/manager-schedule/pom.xml
index f3a4e0a41f..8d598b47ca 100644
--- a/inlong-manager/manager-schedule/pom.xml
+++ b/inlong-manager/manager-schedule/pom.xml
@@ -27,6 +27,7 @@
<artifactId>manager-schedule</artifactId>
<properties>
+ <quartz.version>2.3.2</quartz.version>
<inlong.root.dir>${project.parent.parent.basedir}</inlong.root.dir>
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
</properties>
@@ -37,5 +38,35 @@
<artifactId>manager-pojo</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.quartz-scheduler</groupId>
+ <artifactId>quartz</artifactId>
+ <version>${quartz.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>com.mchange</groupId>
+ <artifactId>c3p0</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.mchange</groupId>
+ <artifactId>mchange-commons-java</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.zaxxer</groupId>
+ <artifactId>HikariCP-java7</artifactId>
+ </exclusion>
+ </exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter</artifactId>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleType.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleType.java
new file mode 100644
index 0000000000..0f296fec42
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleType.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule;
+
+import lombok.Getter;
+
+/**
+ * Type of schedule, including normal and crontab.
+ * */
+@Getter
+public enum ScheduleType {
+
+ NORMAL(0),
+ CRONTAB(1);
+
+ private final int code;
+
+ ScheduleType(int code) {
+ this.code = code;
+ }
+
+ public static ScheduleType fromCode(int code) {
+ for (ScheduleType type : ScheduleType.values()) {
+ if (type.code == code) {
+ return type;
+ }
+ }
+ return null;
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleUnit.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleUnit.java
new file mode 100644
index 0000000000..7b4d60779f
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleUnit.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule;
+
+import lombok.Getter;
+
+@Getter
+public enum ScheduleUnit {
+
+ YEAR("Y"),
+ MONTH("M"),
+ DAY("D"),
+ WEEK("W"),
+ HOUR("H"),
+ MINUTE("I"),
+ SECOND("S"),
+ ONE_WAY("O");
+
+ final String unit;
+
+ ScheduleUnit(String unit) {
+ this.unit = unit;
+ }
+
+ public static ScheduleUnit getScheduleUnit(String unit) {
+ for (ScheduleUnit scheduleUnit : ScheduleUnit.values()) {
+ if (scheduleUnit.unit.equalsIgnoreCase(unit)) {
+ return scheduleUnit;
+ }
+ }
+ return null;
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/exception/QuartzScheduleException.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/exception/QuartzScheduleException.java
new file mode 100644
index 0000000000..2d0ff005bd
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/exception/QuartzScheduleException.java
@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule.exception;
+
+/**
+ * Exceptions occur in the schedule procedure.
+ * */
+public class QuartzScheduleException extends RuntimeException {
+
+ public QuartzScheduleException(String message) {
+ super(message);
+ }
+
+ public QuartzScheduleException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzOfflineSyncJob.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzOfflineSyncJob.java
new file mode 100644
index 0000000000..27628b3320
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzOfflineSyncJob.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule.quartz;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import org.quartz.Job;
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class QuartzOfflineSyncJob implements Job {
+
+ private ScheduleInfo scheduleInfo;
+
+ @Override
+ public void execute(JobExecutionContext context) throws JobExecutionException {
+ // TODO: complete the offline sync logic
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleClient.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleClient.java
new file mode 100644
index 0000000000..41fc8814a1
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleClient.java
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule.quartz;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.schedule.ScheduleEngineClient;
+
+/**
+ * Built-in implementation of schedule engine client corresponding with {@link QuartzScheduleEngine}.
+ * QuartzScheduleClient simply invokes the {@link QuartzScheduleEngine} to register/unregister/update
+ * schedule info instead of calling a remote schedule service.
+ * */
+public class QuartzScheduleClient implements ScheduleEngineClient {
+
+ private final QuartzScheduleEngine scheduleEngine;
+
+ public QuartzScheduleClient(QuartzScheduleEngine scheduleEngine) {
+ this.scheduleEngine = scheduleEngine;
+ }
+
+ @Override
+ public boolean register(ScheduleInfo scheduleInfo) {
+ return scheduleEngine.handleRegister(scheduleInfo);
+ }
+
+ @Override
+ public boolean unregister(ScheduleInfo scheduleInfo) {
+ return scheduleEngine.handleUnregister(scheduleInfo);
+ }
+
+ @Override
+ public boolean update(ScheduleInfo scheduleInfo) {
+ return scheduleEngine.handleUpdate(scheduleInfo);
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngine.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngine.java
new file mode 100644
index 0000000000..31736f1887
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngine.java
@@ -0,0 +1,156 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule.quartz;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.schedule.ScheduleEngine;
+import org.apache.inlong.schedule.exception.QuartzScheduleException;
+
+import com.google.common.annotations.VisibleForTesting;
+import lombok.Getter;
+import org.quartz.JobDetail;
+import org.quartz.JobKey;
+import org.quartz.Scheduler;
+import org.quartz.SchedulerException;
+import org.quartz.Trigger;
+import org.quartz.impl.StdSchedulerFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.inlong.schedule.util.ScheduleUtils.genQuartzJobDetail;
+import static org.apache.inlong.schedule.util.ScheduleUtils.genQuartzTrigger;
+
+/**
+ * The default implementation of schedule engine based on Quartz scheduler. Response for processing
+ * the register/unregister/update requests from {@link QuartzScheduleClient}
+ * */
+@Getter
+public class QuartzScheduleEngine implements ScheduleEngine {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(QuartzScheduleEngine.class);
+
+ private final Scheduler scheduler;
+ private final Set<String> scheduledJobSet = new HashSet<>();
+
+ public QuartzScheduleEngine() {
+ try {
+ this.scheduler = new StdSchedulerFactory().getScheduler();
+ LOGGER.info("Quartz scheduler engine initialized");
+ } catch (SchedulerException e) {
+ throw new QuartzScheduleException("Failed to init quartz scheduler ", e);
+ }
+ }
+
+ @Override
+ public void start() {
+ try {
+ // add listener
+ scheduler.getListenerManager().addSchedulerListener(new QuartzSchedulerListener(this));
+ scheduler.start();
+ LOGGER.info("Quartz scheduler engine started");
+ } catch (SchedulerException e) {
+ throw new QuartzScheduleException("Failed to start quartz scheduler ", e);
+ }
+ }
+
+ /**
+ * Clean job info from scheduledJobSet after trigger finalized.
+ * */
+ public boolean triggerFinalized(Trigger trigger) {
+ String jobName = trigger.getJobKey().getName();
+ LOGGER.info("Trigger finalized for job {}", jobName);
+ return scheduledJobSet.remove(jobName);
+ }
+
+ /**
+ * Handle schedule register.
+ * @param scheduleInfo schedule info to register
+ * */
+ @Override
+ public boolean handleRegister(ScheduleInfo scheduleInfo) {
+ return handleRegister(scheduleInfo, QuartzOfflineSyncJob.class);
+ }
+
+ @VisibleForTesting
+ public boolean handleRegister(ScheduleInfo scheduleInfo, Class<? extends QuartzOfflineSyncJob> clz) {
+ if (scheduledJobSet.contains(scheduleInfo.getInlongGroupId())) {
+ throw new QuartzScheduleException("Group " + scheduleInfo.getInlongGroupId() + " is already registered");
+ }
+ JobDetail jobDetail = genQuartzJobDetail(scheduleInfo, clz);
+ Trigger trigger = genQuartzTrigger(jobDetail, scheduleInfo);
+ try {
+ scheduler.scheduleJob(jobDetail, trigger);
+ scheduledJobSet.add(scheduleInfo.getInlongGroupId());
+ LOGGER.info("Registered new schedule info for {}", scheduleInfo.getInlongGroupId());
+ } catch (SchedulerException e) {
+ throw new QuartzScheduleException(e.getMessage());
+ }
+ return false;
+ }
+
+ /**
+ * Handle schedule unregister.
+ * @param scheduleInfo schedule info to unregister
+ * */
+ @Override
+ public boolean handleUnregister(ScheduleInfo scheduleInfo) {
+ if (scheduledJobSet.contains(scheduleInfo.getInlongGroupId())) {
+ try {
+ scheduler.deleteJob(new JobKey(scheduleInfo.getInlongGroupId()));
+ } catch (SchedulerException e) {
+ throw new QuartzScheduleException(e.getMessage());
+ }
+ }
+ scheduledJobSet.remove(scheduleInfo.getInlongGroupId());
+ LOGGER.info("Un-registered schedule info for {}", scheduleInfo.getInlongGroupId());
+ return true;
+ }
+
+ /**
+ * Handle schedule update.
+ * @param scheduleInfo schedule info to update
+ * */
+ @Override
+ public boolean handleUpdate(ScheduleInfo scheduleInfo) {
+ return handleUpdate(scheduleInfo, QuartzOfflineSyncJob.class);
+ }
+
+ @VisibleForTesting
+ public boolean handleUpdate(ScheduleInfo scheduleInfo, Class<? extends QuartzOfflineSyncJob> clz) {
+ handleUnregister(scheduleInfo);
+ handleRegister(scheduleInfo, clz);
+ LOGGER.info("Updated schedule info for {}", scheduleInfo.getInlongGroupId());
+ return false;
+ }
+
+ @Override
+ public void stop() {
+ if (scheduler != null) {
+ try {
+ scheduler.shutdown();
+ LOGGER.info("Quartz scheduler engine stopped");
+ } catch (SchedulerException e) {
+ throw new QuartzScheduleException("Failed to stop quartz scheduler ", e);
+ }
+ }
+ }
+
+}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzSchedulerListener.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzSchedulerListener.java
new file mode 100644
index 0000000000..ca4f9f1b03
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzSchedulerListener.java
@@ -0,0 +1,137 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule.quartz;
+
+import org.quartz.JobDetail;
+import org.quartz.JobKey;
+import org.quartz.SchedulerException;
+import org.quartz.SchedulerListener;
+import org.quartz.Trigger;
+import org.quartz.TriggerKey;
+
+/**
+ * Default implementation for quartz scheduler listener.
+ * */
+public class QuartzSchedulerListener implements SchedulerListener {
+
+ QuartzScheduleEngine quartzScheduleEngine;
+
+ public QuartzSchedulerListener(QuartzScheduleEngine quartzScheduleEngine) {
+ this.quartzScheduleEngine = quartzScheduleEngine;
+ }
+
+ @Override
+ public void jobScheduled(Trigger trigger) {
+
+ }
+
+ @Override
+ public void jobUnscheduled(TriggerKey triggerKey) {
+
+ }
+
+ @Override
+ public void triggerFinalized(Trigger trigger) {
+ quartzScheduleEngine.triggerFinalized(trigger);
+ }
+
+ @Override
+ public void triggerPaused(TriggerKey triggerKey) {
+
+ }
+
+ @Override
+ public void triggersPaused(String triggerGroup) {
+
+ }
+
+ @Override
+ public void triggerResumed(TriggerKey triggerKey) {
+
+ }
+
+ @Override
+ public void triggersResumed(String triggerGroup) {
+
+ }
+
+ @Override
+ public void jobAdded(JobDetail jobDetail) {
+
+ }
+
+ @Override
+ public void jobDeleted(JobKey jobKey) {
+
+ }
+
+ @Override
+ public void jobPaused(JobKey jobKey) {
+
+ }
+
+ @Override
+ public void jobsPaused(String jobGroup) {
+
+ }
+
+ @Override
+ public void jobResumed(JobKey jobKey) {
+
+ }
+
+ @Override
+ public void jobsResumed(String jobGroup) {
+
+ }
+
+ @Override
+ public void schedulerError(String msg, SchedulerException cause) {
+
+ }
+
+ @Override
+ public void schedulerInStandbyMode() {
+
+ }
+
+ @Override
+ public void schedulerStarted() {
+
+ }
+
+ @Override
+ public void schedulerStarting() {
+
+ }
+
+ @Override
+ public void schedulerShutdown() {
+
+ }
+
+ @Override
+ public void schedulerShuttingdown() {
+
+ }
+
+ @Override
+ public void schedulingDataCleared() {
+
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/util/ScheduleUtils.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/util/ScheduleUtils.java
new file mode 100644
index 0000000000..73114fcb6c
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/util/ScheduleUtils.java
@@ -0,0 +1,139 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule.util;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.schedule.ScheduleType;
+import org.apache.inlong.schedule.ScheduleUnit;
+import org.apache.inlong.schedule.exception.QuartzScheduleException;
+import org.apache.inlong.schedule.quartz.QuartzOfflineSyncJob;
+
+import org.apache.commons.lang3.StringUtils;
+import org.quartz.CronScheduleBuilder;
+import org.quartz.CronTrigger;
+import org.quartz.JobBuilder;
+import org.quartz.JobDetail;
+import org.quartz.ScheduleBuilder;
+import org.quartz.SimpleScheduleBuilder;
+import org.quartz.SimpleTrigger;
+import org.quartz.Trigger;
+import org.quartz.TriggerBuilder;
+
+import java.sql.Timestamp;
+import java.util.Date;
+
+/**
+ * Tools for schedule.
+ * */
+public class ScheduleUtils {
+
+ public static JobDetail genQuartzJobDetail(ScheduleInfo scheduleInfo, Class<? extends QuartzOfflineSyncJob> clz) {
+ return JobBuilder.newJob(clz)
+ .withIdentity(scheduleInfo.getInlongGroupId())
+ .build();
+ }
+
+ public static Trigger genQuartzTrigger(JobDetail jobDetail, ScheduleInfo scheduleInfo) {
+ String key = jobDetail.getKey().getName();
+ Timestamp startTime = scheduleInfo.getStartTime();
+ Timestamp endTime = scheduleInfo.getEndTime();
+ int scheduleType = scheduleInfo.getScheduleType();
+ ScheduleType type = ScheduleType.fromCode(scheduleType);
+ if (type == null) {
+ throw new QuartzScheduleException("Invalid schedule type: " + scheduleType);
+ }
+ switch (type) {
+ case NORMAL:
+ return TriggerBuilder.newTrigger()
+ .withIdentity(key)
+ .startAt(new Date(startTime.getTime()))
+ .endAt(new Date(endTime.getTime()))
+ .withSchedule(genSimpleQuartzScheduleBuilder(scheduleInfo.getScheduleInterval(),
+ scheduleInfo.getScheduleUnit()))
+ .forJob(jobDetail).build();
+ case CRONTAB:
+ return TriggerBuilder.newTrigger()
+ .withIdentity(key)
+ .startAt(new Date(startTime.getTime()))
+ .endAt(new Date(endTime.getTime()))
+ .withSchedule(genCronQuartzScheduleBuilder(scheduleInfo.getCrontabExpression()))
+ .forJob(jobDetail).build();
+ default:
+ throw new QuartzScheduleException("Unknown schedule type: " + scheduleType);
+ }
+ }
+
+ // Y=year, M=month, W=week, D=day, H=hour, I=minute, O=oneway
+ public static ScheduleBuilder<SimpleTrigger> genSimpleQuartzScheduleBuilder(int interval, String scheduleUnit) {
+ if (StringUtils.isBlank(scheduleUnit)) {
+ throw new QuartzScheduleException("Schedule unit cannot be empty");
+ }
+ ScheduleUnit unit = ScheduleUnit.getScheduleUnit(scheduleUnit);
+ if (unit == null) {
+ throw new QuartzScheduleException("Unknown schedule unit: " + scheduleUnit);
+ }
+ switch (unit) {
+ case YEAR:
+ return SimpleScheduleBuilder
+ .simpleSchedule()
+ .withIntervalInHours(365 * 24 * interval)
+ .repeatForever();
+ case MONTH:
+ return SimpleScheduleBuilder
+ .simpleSchedule()
+ .withIntervalInHours(30 * 24 * interval)
+ .repeatForever();
+ case WEEK:
+ return SimpleScheduleBuilder
+ .simpleSchedule()
+ .withIntervalInHours(7 * 24 * interval)
+ .repeatForever();
+ case DAY:
+ return SimpleScheduleBuilder
+ .simpleSchedule()
+ .withIntervalInHours(24 * interval)
+ .repeatForever();
+ case HOUR:
+ return SimpleScheduleBuilder
+ .simpleSchedule()
+ .withIntervalInHours(interval)
+ .repeatForever();
+ case MINUTE:
+ return SimpleScheduleBuilder
+ .simpleSchedule()
+ .withIntervalInMinutes(interval)
+ .repeatForever();
+ case SECOND:
+ return SimpleScheduleBuilder
+ .simpleSchedule()
+ .withIntervalInSeconds(interval)
+ .repeatForever();
+ case ONE_WAY:
+ return SimpleScheduleBuilder
+ .simpleSchedule()
+ .withIntervalInSeconds(interval)
+ .withRepeatCount(1);
+ default:
+ throw new QuartzScheduleException("Not supported schedule interval" + scheduleUnit);
+ }
+ }
+
+ public static ScheduleBuilder<CronTrigger> genCronQuartzScheduleBuilder(String cronExpression) {
+ return CronScheduleBuilder.cronSchedule(cronExpression).withMisfireHandlingInstructionIgnoreMisfires();
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/BaseScheduleTest.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/BaseScheduleTest.java
new file mode 100644
index 0000000000..c0ccc4b14c
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/BaseScheduleTest.java
@@ -0,0 +1,116 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.schedule.exception.QuartzScheduleException;
+
+import java.sql.Timestamp;
+
+import static org.apache.inlong.schedule.ScheduleUnit.SECOND;
+
+public class BaseScheduleTest {
+
+ public static final int SCHEDULE_TYPE_NORMAL = 0;
+ public static final int SCHEDULE_TYPE_CRON = 1;
+ public static final int DEFAULT_INTERVAL = 2;
+ public static final long DEFAULT_SPAN_IN_MS = 10 * 1000;
+ public static final String ILLEGAL_TIMEUNIT = "I";
+ public static final String GROUP_ID = "testGroup";
+ public static final String CRON_EXPRESSION_PER_SECONDS = "*/1 * * * * ?";
+ public static final int CRON_SCHEDULE_INTERVAL_PER_SECONDS = 1;
+ public static final String CRON_EXPRESSION_EVERY_TWO_SECONDS = "*/2 * * * * ?";
+ public static final int CRON_SCHEDULE_INTERVAL_EVERY_TWO_SECONDS = 2;
+ public static final String ILLEGAL_CRON_EXPRESSION = "*/1 * * ?";
+
+ public ScheduleInfo genDefaultScheduleInfo() {
+ return genNormalScheduleInfo(GROUP_ID, SECOND.getUnit(), DEFAULT_INTERVAL, DEFAULT_SPAN_IN_MS);
+ }
+
+ public ScheduleInfo genNormalScheduleInfo(String groupId, String scheduleUnit, int scheduleInterval,
+ long timeSpanInMs) {
+ ScheduleInfo scheduleInfo = new ScheduleInfo();
+ scheduleInfo.setInlongGroupId(groupId);
+ scheduleInfo.setScheduleType(SCHEDULE_TYPE_NORMAL);
+ scheduleInfo.setScheduleUnit(scheduleUnit);
+ scheduleInfo.setScheduleInterval(scheduleInterval);
+ setStartAndEndTime(scheduleInfo, timeSpanInMs);
+ return scheduleInfo;
+ }
+
+ public ScheduleInfo genDefaultCronScheduleInfo() {
+ return genCronScheduleInfo(GROUP_ID, CRON_EXPRESSION_PER_SECONDS, DEFAULT_SPAN_IN_MS);
+ }
+
+ public ScheduleInfo genCronScheduleInfo(String groupId, String cronExpression, long timeSpanInMs) {
+ ScheduleInfo scheduleInfo = new ScheduleInfo();
+ scheduleInfo.setInlongGroupId(groupId);
+ scheduleInfo.setScheduleType(SCHEDULE_TYPE_CRON);
+ scheduleInfo.setCrontabExpression(cronExpression);
+ setStartAndEndTime(scheduleInfo, timeSpanInMs);
+ return scheduleInfo;
+ }
+
+ private void setStartAndEndTime(ScheduleInfo scheduleInfo, long timeSpanInMs) {
+ long startTime = System.currentTimeMillis() / 1000 * 1000;
+ long endTime = startTime + timeSpanInMs;
+ scheduleInfo.setStartTime(new Timestamp(startTime));
+ scheduleInfo.setEndTime(new Timestamp(endTime));
+ }
+
+ protected long calculateScheduleTimes(ScheduleInfo scheduleInfo, boolean isCron) {
+
+ long timeSpanInMs = scheduleInfo.getEndTime().getTime() - scheduleInfo.getStartTime().getTime();
+ int interval = -1;
+ ScheduleUnit scheduleUnit = null;
+ if (isCron) {
+ if (scheduleInfo.getCrontabExpression().equalsIgnoreCase(CRON_EXPRESSION_PER_SECONDS)) {
+ interval = CRON_SCHEDULE_INTERVAL_PER_SECONDS;
+ } else if (scheduleInfo.getCrontabExpression().equalsIgnoreCase(CRON_EXPRESSION_EVERY_TWO_SECONDS)) {
+ interval = CRON_SCHEDULE_INTERVAL_EVERY_TWO_SECONDS;
+ }
+ scheduleUnit = SECOND;
+ } else {
+ interval = scheduleInfo.getScheduleInterval();
+ scheduleUnit = ScheduleUnit.getScheduleUnit(scheduleInfo.getScheduleUnit());
+ }
+ if (scheduleUnit == null) {
+ throw new QuartzScheduleException("Schedule unit is null");
+ }
+ switch (scheduleUnit) {
+ case YEAR:
+ return timeSpanInMs / 365 / 1000 / 3600 / 24 / 7 / interval;
+ case MONTH:
+ return timeSpanInMs / 30 / 1000 / 3600 / 24 / 7 / interval;
+ case WEEK:
+ return timeSpanInMs / 1000 / 3600 / 24 / 7 / interval;
+ case DAY:
+ return timeSpanInMs / 1000 / 3600 / 24 / interval;
+ case HOUR:
+ return timeSpanInMs / 1000 / 3600 / interval;
+ case MINUTE:
+ return timeSpanInMs / 1000 / 60 / interval;
+ case SECOND:
+ return timeSpanInMs / 1000 / interval;
+ case ONE_WAY:
+ return 1;
+ default:
+ return 0;
+ }
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/MockJob.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/MockJob.java
new file mode 100644
index 0000000000..47ce19ad09
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/MockJob.java
@@ -0,0 +1,48 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule.quartz;
+
+import org.quartz.JobExecutionContext;
+import org.quartz.JobExecutionException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class MockJob extends QuartzOfflineSyncJob {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(MockJob.class);
+
+ public static CountDownLatch countDownLatch;
+ private static AtomicInteger counter = new AtomicInteger(0);
+
+ @Override
+ public void execute(JobExecutionContext context) throws JobExecutionException {
+ if (countDownLatch.getCount() > 0) {
+ countDownLatch.countDown();
+ }
+ LOGGER.info("MockJob executed " + counter.incrementAndGet());
+ }
+
+ public static void setCount(int count) {
+ countDownLatch = new CountDownLatch(count);
+ counter.set(0);
+ LOGGER.info("MockJob has been reset.");
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngineTest.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngineTest.java
new file mode 100644
index 0000000000..fc5ca9d9fd
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngineTest.java
@@ -0,0 +1,161 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule.quartz;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.schedule.BaseScheduleTest;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.quartz.JobKey;
+
+import java.util.concurrent.TimeUnit;
+
+import static org.apache.inlong.schedule.ScheduleUnit.SECOND;
+import static org.awaitility.Awaitility.await;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class QuartzScheduleEngineTest extends BaseScheduleTest {
+
+ private static QuartzScheduleEngine scheduleEngine;
+
+ @BeforeAll
+ public static void initScheduleEngine() throws Exception {
+ scheduleEngine = new QuartzScheduleEngine();
+ scheduleEngine.start();
+ }
+
+ @Test
+ @Timeout(30)
+ public void testRegisterScheduleInfo() throws Exception {
+ // 1. test for normal schedule
+ ScheduleInfo scheduleInfo = genDefaultScheduleInfo();
+ testRegister(scheduleInfo, false);
+
+ // 2. test for cron schedule
+ scheduleInfo = genDefaultCronScheduleInfo();
+ testRegister(scheduleInfo, true);
+ }
+
+ private void testRegister(ScheduleInfo scheduleInfo, boolean isCrontab) throws Exception {
+ // cal total schedule times
+ long expectCount = calculateScheduleTimes(scheduleInfo, isCrontab);
+ // set countdown latch
+ MockJob.setCount((int) expectCount);
+ // register schedule info
+ scheduleEngine.handleRegister(scheduleInfo, MockJob.class);
+ // check job exist
+ assertEquals(1, scheduleEngine.getScheduledJobSet().size());
+ JobKey jobKey = new JobKey(scheduleInfo.getInlongGroupId());
+ boolean exist = scheduleEngine.getScheduler().checkExists(jobKey);
+ assertTrue(exist);
+ MockJob.countDownLatch.await();
+
+ // not job exist after scheduled
+ await().atMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(0, scheduleEngine.getScheduledJobSet().size());
+ assertFalse(scheduleEngine.getScheduler().checkExists(jobKey));
+ });
+ }
+
+ @Test
+ @Timeout(30)
+ public void testUnRegisterScheduleInfo() throws Exception {
+ // 1. test for normal schedule
+ ScheduleInfo scheduleInfo = genDefaultScheduleInfo();
+ testUnRegister(scheduleInfo, false);
+
+ // 2. test for cron schedule, gen cron schedule info, */2 * * * * ?
+ scheduleInfo = genDefaultCronScheduleInfo();
+ testUnRegister(scheduleInfo, true);
+ }
+
+ private void testUnRegister(ScheduleInfo scheduleInfo, boolean isCrontab) throws Exception {
+ // cal total schedule times
+ long expectCount = calculateScheduleTimes(scheduleInfo, isCrontab);
+
+ MockJob.setCount((int) (expectCount / 2));
+ // register schedule info
+ scheduleEngine.handleRegister(scheduleInfo, MockJob.class);
+ // check job exist
+ assertEquals(1, scheduleEngine.getScheduledJobSet().size());
+ JobKey jobKey = new JobKey(scheduleInfo.getInlongGroupId());
+ boolean exist = scheduleEngine.getScheduler().checkExists(jobKey);
+ assertTrue(exist);
+ MockJob.countDownLatch.await();
+
+ // un-register before trigger finalized
+ scheduleEngine.handleUnregister(scheduleInfo);
+ // not job exist after un-register
+ assertEquals(0, scheduleEngine.getScheduledJobSet().size());
+ exist = scheduleEngine.getScheduler().checkExists(jobKey);
+ assertFalse(exist);
+ }
+
+ @Test
+ @Timeout(50)
+ public void testUpdateScheduleInfo() throws Exception {
+ // 1. test for normal schedule
+ ScheduleInfo scheduleInfo =
+ genNormalScheduleInfo(GROUP_ID, SECOND.getUnit(), DEFAULT_INTERVAL, DEFAULT_SPAN_IN_MS);
+ ScheduleInfo scheduleInfoToUpdate =
+ genNormalScheduleInfo(GROUP_ID, SECOND.getUnit(), DEFAULT_INTERVAL / 2, DEFAULT_SPAN_IN_MS);
+ testUpdate(scheduleInfo, scheduleInfoToUpdate, false);
+
+ // 2. test for cron schedule
+ scheduleInfo = genCronScheduleInfo(GROUP_ID, CRON_EXPRESSION_EVERY_TWO_SECONDS, DEFAULT_SPAN_IN_MS);
+ scheduleInfoToUpdate = genCronScheduleInfo(GROUP_ID, CRON_EXPRESSION_PER_SECONDS, DEFAULT_SPAN_IN_MS);
+ testUpdate(scheduleInfo, scheduleInfoToUpdate, true);
+ }
+
+ public void testUpdate(ScheduleInfo scheduleInfo, ScheduleInfo scheduleInfoToUpdate, boolean isCrontab)
+ throws Exception {
+ // cal total schedule times
+ long expectCount = calculateScheduleTimes(scheduleInfo, isCrontab);
+ MockJob.setCount((int) (expectCount / 2));
+ // register schedule info
+ scheduleEngine.handleRegister(scheduleInfo, MockJob.class);
+ // check job exist
+ assertEquals(1, scheduleEngine.getScheduledJobSet().size());
+ JobKey jobKey = new JobKey(scheduleInfo.getInlongGroupId());
+ boolean exist = scheduleEngine.getScheduler().checkExists(jobKey);
+ assertTrue(exist);
+ MockJob.countDownLatch.await();
+
+ // update schedule before trigger finalized
+ expectCount = calculateScheduleTimes(scheduleInfoToUpdate, isCrontab);
+ MockJob.setCount((int) expectCount);
+ scheduleEngine.handleUpdate(scheduleInfoToUpdate, MockJob.class);
+
+ // job scheduled after updated
+ assertEquals(1, scheduleEngine.getScheduledJobSet().size());
+ exist = scheduleEngine.getScheduler().checkExists(jobKey);
+ assertTrue(exist);
+
+ MockJob.countDownLatch.await();
+
+ // not job exist after scheduled
+ await().atMost(3, TimeUnit.SECONDS).untilAsserted(() -> {
+ assertEquals(0, scheduleEngine.getScheduledJobSet().size());
+ assertFalse(scheduleEngine.getScheduler().checkExists(jobKey));
+ });
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/util/ScheduleUtilsTest.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/util/ScheduleUtilsTest.java
new file mode 100644
index 0000000000..77b84f3064
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/util/ScheduleUtilsTest.java
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule.util;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.schedule.BaseScheduleTest;
+import org.apache.inlong.schedule.exception.QuartzScheduleException;
+import org.apache.inlong.schedule.quartz.QuartzOfflineSyncJob;
+
+import org.junit.jupiter.api.Test;
+import org.quartz.CronScheduleBuilder;
+import org.quartz.CronTrigger;
+import org.quartz.JobDetail;
+import org.quartz.JobKey;
+import org.quartz.ScheduleBuilder;
+import org.quartz.SimpleScheduleBuilder;
+import org.quartz.SimpleTrigger;
+import org.quartz.Trigger;
+import org.quartz.TriggerKey;
+
+import java.util.Date;
+
+import static org.apache.inlong.schedule.ScheduleUnit.DAY;
+import static org.apache.inlong.schedule.ScheduleUnit.HOUR;
+import static org.apache.inlong.schedule.ScheduleUnit.MINUTE;
+import static org.apache.inlong.schedule.ScheduleUnit.MONTH;
+import static org.apache.inlong.schedule.ScheduleUnit.ONE_WAY;
+import static org.apache.inlong.schedule.ScheduleUnit.WEEK;
+import static org.apache.inlong.schedule.ScheduleUnit.YEAR;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class ScheduleUtilsTest extends BaseScheduleTest {
+
+ @Test
+ public void testGenScheduleBuilder() {
+ ScheduleBuilder<SimpleTrigger> builder =
+ ScheduleUtils.genSimpleQuartzScheduleBuilder(DEFAULT_INTERVAL, YEAR.getUnit());
+ assertNotNull(builder);
+ assertInstanceOf(SimpleScheduleBuilder.class, builder);
+
+ builder = ScheduleUtils.genSimpleQuartzScheduleBuilder(DEFAULT_INTERVAL, MONTH.getUnit());
+ assertNotNull(builder);
+ assertInstanceOf(SimpleScheduleBuilder.class, builder);
+
+ builder = ScheduleUtils.genSimpleQuartzScheduleBuilder(DEFAULT_INTERVAL, WEEK.getUnit());
+ assertNotNull(builder);
+ assertInstanceOf(SimpleScheduleBuilder.class, builder);
+
+ builder = ScheduleUtils.genSimpleQuartzScheduleBuilder(DEFAULT_INTERVAL, DAY.getUnit());
+ assertNotNull(builder);
+ assertInstanceOf(SimpleScheduleBuilder.class, builder);
+
+ builder = ScheduleUtils.genSimpleQuartzScheduleBuilder(DEFAULT_INTERVAL, HOUR.getUnit());
+ assertNotNull(builder);
+ assertInstanceOf(SimpleScheduleBuilder.class, builder);
+
+ builder = ScheduleUtils.genSimpleQuartzScheduleBuilder(DEFAULT_INTERVAL, MINUTE.getUnit());
+ assertNotNull(builder);
+ assertInstanceOf(SimpleScheduleBuilder.class, builder);
+
+ builder = ScheduleUtils.genSimpleQuartzScheduleBuilder(DEFAULT_INTERVAL, ONE_WAY.getUnit());
+ assertNotNull(builder);
+ assertInstanceOf(SimpleScheduleBuilder.class, builder);
+
+ try {
+ ScheduleUtils.genSimpleQuartzScheduleBuilder(DEFAULT_INTERVAL, ILLEGAL_TIMEUNIT);
+ } catch (Exception e) {
+ assertInstanceOf(QuartzScheduleException.class, e);
+ }
+
+ ScheduleBuilder<CronTrigger> cronBuilder = ScheduleUtils.genCronQuartzScheduleBuilder(
+ CRON_EXPRESSION_EVERY_TWO_SECONDS);
+ assertNotNull(cronBuilder);
+ assertInstanceOf(CronScheduleBuilder.class, cronBuilder);
+
+ try {
+ ScheduleUtils.genCronQuartzScheduleBuilder(ILLEGAL_CRON_EXPRESSION);
+ } catch (Exception e) {
+ String errorMsg = e.getMessage();
+ assertTrue(errorMsg.contains(ILLEGAL_CRON_EXPRESSION));
+ }
+ }
+
+ @Test
+ public void testGenJobDetail() {
+ ScheduleInfo scheduleInfo = genDefaultScheduleInfo();
+ JobDetail jobDetail = ScheduleUtils.genQuartzJobDetail(scheduleInfo, QuartzOfflineSyncJob.class);
+ assertNotNull(jobDetail);
+
+ JobKey jobKey = jobDetail.getKey();
+ assertNotNull(jobKey);
+
+ String identity = jobKey.getName();
+ assertEquals(scheduleInfo.getInlongGroupId(), identity);
+ }
+
+ @Test
+ public void testGenCronTrigger() {
+ // normal
+ ScheduleInfo scheduleInfo = genDefaultScheduleInfo();
+ JobDetail jobDetail = ScheduleUtils.genQuartzJobDetail(scheduleInfo, QuartzOfflineSyncJob.class);
+
+ Trigger trigger = ScheduleUtils.genQuartzTrigger(jobDetail, scheduleInfo);
+ assertNotNull(trigger);
+
+ TriggerKey triggerKey = trigger.getKey();
+ assertNotNull(triggerKey);
+ String identity = triggerKey.getName();
+ assertEquals(scheduleInfo.getInlongGroupId(), identity);
+
+ ScheduleBuilder<? extends Trigger> scheduleBuilder = trigger.getScheduleBuilder();
+ assertInstanceOf(SimpleScheduleBuilder.class, scheduleBuilder);
+
+ Date startDate = trigger.getStartTime();
+ assertNotNull(startDate);
+ assertEquals(startDate.getTime(), scheduleInfo.getStartTime().getTime());
+
+ Date endDate = trigger.getEndTime();
+ assertNotNull(endDate);
+ assertEquals(endDate.getTime(), scheduleInfo.getEndTime().getTime());
+
+ // cron
+ scheduleInfo = genDefaultCronScheduleInfo();
+ jobDetail = ScheduleUtils.genQuartzJobDetail(scheduleInfo, QuartzOfflineSyncJob.class);
+
+ trigger = ScheduleUtils.genQuartzTrigger(jobDetail, scheduleInfo);
+ assertNotNull(trigger);
+
+ triggerKey = trigger.getKey();
+ assertNotNull(triggerKey);
+ identity = triggerKey.getName();
+ assertEquals(scheduleInfo.getInlongGroupId(), identity);
+
+ scheduleBuilder = trigger.getScheduleBuilder();
+ assertInstanceOf(CronScheduleBuilder.class, scheduleBuilder);
+
+ startDate = trigger.getStartTime();
+ assertNotNull(startDate);
+ assertEquals(startDate.getTime(), scheduleInfo.getStartTime().getTime());
+
+ endDate = trigger.getEndTime();
+ assertNotNull(endDate);
+ assertEquals(endDate.getTime(), scheduleInfo.getEndTime().getTime());
+
+ }
+}
diff --git a/inlong-manager/manager-schedule/src/test/resources/log4j2.xml b/inlong-manager/manager-schedule/src/test/resources/log4j2.xml
new file mode 100644
index 0000000000..8f2da663e7
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/test/resources/log4j2.xml
@@ -0,0 +1,46 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you 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
+
+ http://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.
+-->
+<configuration status="WARN" monitorInterval="30">
+ <Properties>
+ <property name="basePath">logs</property>
+ <property name="log_pattern">%d{yyyy-MM-dd HH:mm:ss.SSS} -%5p [%5.30t] %-30.30C{1.}:%L - %m%n</property>
+ <property name="output_log_level">DEBUG</property>
+ <property name="all_fileName">${basePath}/manager-service-ut.log</property>
+ <property name="console_print_level">DEBUG</property>
+ </Properties>
+
+ <appenders>
+ <Console name="Console" target="SYSTEM_OUT">
+ <ThresholdFilter level="${console_print_level}" onMatch="ACCEPT" onMismatch="DENY"/>
+ <PatternLayout pattern="${log_pattern}"/>
+ <follow>true</follow>
+ </Console>
+ <File name="AllFile" fileName="${all_fileName}">
+ <PatternLayout pattern="${log_pattern}"/>
+ </File>
+ </appenders>
+
+ <loggers>
+ <root level="${output_log_level}">
+ <appender-ref ref="Console"/>
+ <appender-ref ref="AllFile"/>
+ </root>
+ </loggers>
+</configuration>
\ No newline at end of file
diff --git a/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql b/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql
index a8b545f7bf..babe5139f8 100644
--- a/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql
+++ b/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql
@@ -970,7 +970,7 @@ CREATE TABLE IF NOT EXISTS `schedule_config`
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, undeleted ones cannot be repeated',
`schedule_type` int(4) NOT NULL DEFAULT '0' COMMENT 'Schedule type, 0 for normal, 1 for crontab',
- `schedule_unit` varchar(64) NOT NULL COMMENT 'Schedule unit,M=month, W=week, D=day, H=hour, M=minute, O=oneway',
+ `schedule_unit` varchar(64) NOT NULL COMMENT 'Schedule unit, Y=year, M=month, W=week, D=day, H=hour, I=minute, O=oneway',
`schedule_interval` int(11) DEFAULT '1' COMMENT 'Schedule interval',
`start_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Start time for schedule',
`end_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'End time for schedule',
diff --git a/inlong-manager/manager-web/sql/apache_inlong_manager.sql b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
index f7dc1c0a90..b4458af76f 100644
--- a/inlong-manager/manager-web/sql/apache_inlong_manager.sql
+++ b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
@@ -1022,7 +1022,7 @@ CREATE TABLE IF NOT EXISTS `schedule_config`
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, undeleted ones cannot be repeated',
`schedule_type` int(4) NOT NULL DEFAULT '0' COMMENT 'Schedule type, 0 for normal, 1 for crontab',
- `schedule_unit` varchar(64) NOT NULL COMMENT 'Schedule unit,M=month, W=week, D=day, H=hour, M=minute, O=oneway',
+ `schedule_unit` varchar(64) NOT NULL COMMENT 'Schedule unit, Y=year, M=month, W=week, D=day, H=hour, I=minute, O=oneway',
`schedule_interval` int(11) DEFAULT '1' COMMENT 'Schedule interval',
`start_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Start time for schedule',
`end_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'End time for schedule',
diff --git a/inlong-manager/manager-web/sql/changes-1.13.0.sql b/inlong-manager/manager-web/sql/changes-1.13.0.sql
index 38e7627c5e..6a248ef0a2 100644
--- a/inlong-manager/manager-web/sql/changes-1.13.0.sql
+++ b/inlong-manager/manager-web/sql/changes-1.13.0.sql
@@ -97,7 +97,7 @@ CREATE TABLE IF NOT EXISTS `schedule_config`
`id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
`inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, undeleted ones cannot be repeated',
`schedule_type` int(4) NOT NULL DEFAULT '0' COMMENT 'Schedule type, 0 for normal, 1 for crontab',
- `schedule_unit` varchar(64) NOT NULL COMMENT 'Schedule unit,M=month, W=week, D=day, H=hour, M=minute, O=oneway',
+ `schedule_unit` varchar(64) NOT NULL COMMENT 'Schedule unit, Y=year, M=month, W=week, D=day, H=hour, I=minute, O=oneway',
`schedule_interval` int(11) DEFAULT '1' COMMENT 'Schedule interval',
`start_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Start time for schedule',
`end_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'End time for schedule',
diff --git a/inlong-manager/pom.xml b/inlong-manager/pom.xml
index bff220728b..cc7421d1c1 100644
--- a/inlong-manager/pom.xml
+++ b/inlong-manager/pom.xml
@@ -40,8 +40,8 @@
<module>manager-service</module>
<module>manager-workflow</module>
<module>manager-web</module>
- <module>manager-docker</module>
<module>manager-schedule</module>
+ <module>manager-docker</module>
</modules>
<properties>
(inlong) 03/09: [INLONG-9822][Manager] Support flink job runtime execution mode configuration (#9823)
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit b83a4831d419435858e5d5b2a538fc8fbdcc2333
Author: AloysZhang <al...@apache.org>
AuthorDate: Fri Mar 15 15:12:53 2024 +0800
[INLONG-9822][Manager] Support flink job runtime execution mode configuration (#9823)
---
.../apache/inlong/manager/common/consts/InlongConstants.java | 3 +++
.../org/apache/inlong/manager/plugin/flink/FlinkService.java | 2 ++
.../org/apache/inlong/manager/plugin/flink/dto/FlinkInfo.java | 2 ++
.../inlong/manager/plugin/listener/DeleteSortListener.java | 1 +
.../inlong/manager/plugin/listener/RestartSortListener.java | 8 +++++++-
.../java/org/apache/inlong/sort/configuration/Constants.java | 8 ++++++++
.../src/main/java/org/apache/inlong/sort/Entrance.java | 11 ++++++++++-
7 files changed, 33 insertions(+), 2 deletions(-)
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java
index 6ef9f5772c..3438001314 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java
@@ -90,6 +90,9 @@ public class InlongConstants {
public static final Integer DATASYNC_REALTIME_MODE = 1;
public static final Integer DATASYNC_OFFLINE_MODE = 2;
+ public static final String RUNTIME_EXECUTION_MODE_STREAMING = "streaming";
+ public static final String RUNTIME_EXECUTION_MODE_BATCH = "batch";
+
public static final Integer DISABLE_ZK = 0;
public static final Integer ENABLE_ZK = 1;
diff --git a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java
index 129b7aa444..9188c6b73f 100644
--- a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java
+++ b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/FlinkService.java
@@ -258,6 +258,8 @@ public class FlinkService {
list.add(flinkInfo.getLocalConfPath());
list.add("-checkpoint.interval");
list.add("60000");
+ list.add("-runtime.execution.mode");
+ list.add(flinkInfo.getRuntimeExecutionMode());
return list.toArray(new String[0]);
}
diff --git a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/dto/FlinkInfo.java b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/dto/FlinkInfo.java
index f7071ceb4b..4c3c75f855 100644
--- a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/dto/FlinkInfo.java
+++ b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/flink/dto/FlinkInfo.java
@@ -52,4 +52,6 @@ public class FlinkInfo {
private boolean isException = false;
private String exceptionMsg;
+
+ private String runtimeExecutionMode;
}
diff --git a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java
index 009374e2a6..5e30ad8cb5 100644
--- a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java
+++ b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java
@@ -119,6 +119,7 @@ public class DeleteSortListener implements SortOperateListener {
FlinkOperation flinkOperation = FlinkOperation.getInstance();
try {
flinkOperation.delete(flinkInfo);
+ // TODO if the job is OFFLINE, should delete the scheduler information
log.info("job delete success for jobId={}", jobId);
} catch (Exception e) {
flinkInfo.setException(true);
diff --git a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java
index 0c6828c984..242138c1e4 100644
--- a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java
+++ b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java
@@ -86,7 +86,13 @@ public class RestartSortListener implements SortOperateListener {
}
GroupResourceProcessForm groupResourceProcessForm = (GroupResourceProcessForm) processForm;
-
+ if (InlongConstants.DATASYNC_OFFLINE_MODE.equals(
+ groupResourceProcessForm.getGroupInfo().getInlongGroupMode())) {
+ String message = String.format("offline data sync job should be scheduled by the "
+ + "scheduler system only for groupId [%s]", groupId);
+ log.error(message);
+ return ListenerResult.fail(message);
+ }
List<InlongStreamInfo> streamInfos = groupResourceProcessForm.getStreamInfos();
for (InlongStreamInfo streamInfo : streamInfos) {
List<StreamSink> sinkList = streamInfo.getSinkList();
diff --git a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/Constants.java b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
index 702dbef53f..a5717242a5 100644
--- a/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
+++ b/inlong-sort/sort-common/src/main/java/org/apache/inlong/sort/configuration/Constants.java
@@ -291,6 +291,14 @@ public class Constants {
public static final ConfigOption<Integer> CHECKPOINT_TIMEOUT_MS =
key("checkpoint.timeout.ms").defaultValue(600000);
+ // ------------------------------------------------------------------------
+ // Flink runtime execution mode, including stream and batch, default is stream
+ // ------------------------------------------------------------------------
+
+ public static final ConfigOption<String> RUNTIME_EXECUTION_MODE = key("runtime.execution.mode")
+ .defaultValue("stream")
+ .withDescription("The runtime execution mode of Flink, including stream and batch, default is stream");
+
// ------------------------------------------------------------------------
// Metrics related
// ------------------------------------------------------------------------
diff --git a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/Entrance.java b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/Entrance.java
index ea3f3f95be..4aa3902cae 100644
--- a/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/Entrance.java
+++ b/inlong-sort/sort-core/src/main/java/org/apache/inlong/sort/Entrance.java
@@ -40,6 +40,8 @@ import java.nio.charset.StandardCharsets;
public class Entrance {
+ public static final String BATCH_MODE = "batch";
+
public static void main(String[] args) throws Exception {
final ParameterTool parameterTool = ParameterTool.fromArgs(args);
final Configuration config = parameterTool.getConfiguration();
@@ -50,7 +52,14 @@ public class Entrance {
config.getInteger(Constants.MIN_PAUSE_BETWEEN_CHECKPOINTS_MS));
env.getCheckpointConfig().setCheckpointTimeout(config.getInteger(Constants.CHECKPOINT_TIMEOUT_MS));
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
- EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
+
+ String runtimeExecutionMode = config.getString(Constants.RUNTIME_EXECUTION_MODE);
+ EnvironmentSettings settings;
+ if (BATCH_MODE.equalsIgnoreCase(runtimeExecutionMode)) {
+ settings = EnvironmentSettings.newInstance().inBatchMode().build();
+ } else {
+ settings = EnvironmentSettings.newInstance().inStreamingMode().build();
+ }
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);
tableEnv.getConfig().getConfiguration().setString(Constants.PIPELINE_NAME,
config.getString(Constants.JOB_NAME));
(inlong) 04/09: [INLONG-9862][Manager] Support submit flink job for offline sync (#9865)
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit dc8aa2ef5e57428348fa49da78feb2543c951252
Author: AloysZhang <al...@apache.org>
AuthorDate: Fri Mar 22 18:24:26 2024 +0800
[INLONG-9862][Manager] Support submit flink job for offline sync (#9865)
---
.../manager/common/consts/InlongConstants.java | 4 +
.../manager/pojo/sort/util/StreamParseUtils.java | 21 +++++
.../listener/StreamTaskListenerFactory.java | 28 +++++++
.../schedule/StreamScheduleResourceListener.java | 95 ++++++++++++++++++++++
.../service/listener/sort/SortConfigListener.java | 8 ++
.../stream/CreateStreamWorkflowDefinition.java | 13 ++-
.../workflow/definition/ServiceTaskType.java | 1 +
.../task/ScheduleOperateListener.java} | 36 ++++----
.../manager/workflow/plugin/ProcessPlugin.java | 5 ++
9 files changed, 188 insertions(+), 23 deletions(-)
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java
index 3438001314..e9a18439d2 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java
@@ -143,6 +143,10 @@ public class InlongConstants {
*/
public static final String DATAFLOW = "dataflow";
+ public static final String REGISTER_SCHEDULE_STATUS = "register.schedule.status";
+
+ public static final String REGISTERED = "registered";
+
public static final String STREAMS = "streams";
public static final String RELATIONS = "relations";
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/StreamParseUtils.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/StreamParseUtils.java
index 92470968ac..203bcc6778 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/StreamParseUtils.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/StreamParseUtils.java
@@ -17,11 +17,14 @@
package org.apache.inlong.manager.pojo.sort.util;
+import org.apache.inlong.manager.common.consts.InlongConstants;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.TransformType;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.pojo.sink.StreamSink;
import org.apache.inlong.manager.pojo.source.StreamSource;
+import org.apache.inlong.manager.pojo.stream.InlongStreamExtInfo;
+import org.apache.inlong.manager.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.pojo.stream.StreamNode;
import org.apache.inlong.manager.pojo.stream.StreamPipeline;
import org.apache.inlong.manager.pojo.stream.StreamTransform;
@@ -38,6 +41,8 @@ import org.apache.inlong.manager.pojo.transform.splitter.SplitterDefinition;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
+import org.apache.commons.collections4.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
/**
* Utils of stream parse.
@@ -154,4 +159,20 @@ public class StreamParseUtils {
return GSON.fromJson(tempView, StreamPipeline.class);
}
+ public static String getStreamExtProperty(String key, InlongStreamInfo streamInfo) {
+ if (StringUtils.isNotBlank(key) && streamInfo != null && CollectionUtils.isNotEmpty(streamInfo.getExtList())) {
+ for (InlongStreamExtInfo ext : streamInfo.getExtList()) {
+ if (key.equalsIgnoreCase(ext.getKeyName())) {
+ return ext.getKeyValue();
+ }
+ }
+ }
+ return null;
+ }
+
+ public static boolean isRegisterScheduleSuccess(InlongStreamInfo streamInfo) {
+ return InlongConstants.REGISTERED
+ .equalsIgnoreCase(getStreamExtProperty(InlongConstants.REGISTER_SCHEDULE_STATUS, streamInfo));
+ }
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/StreamTaskListenerFactory.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/StreamTaskListenerFactory.java
index 43fdfe4666..87c3a867c2 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/StreamTaskListenerFactory.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/StreamTaskListenerFactory.java
@@ -20,12 +20,14 @@ package org.apache.inlong.manager.service.listener;
import org.apache.inlong.manager.common.plugin.Plugin;
import org.apache.inlong.manager.common.plugin.PluginBinder;
import org.apache.inlong.manager.service.listener.queue.StreamQueueResourceListener;
+import org.apache.inlong.manager.service.listener.schedule.StreamScheduleResourceListener;
import org.apache.inlong.manager.service.listener.sink.StreamSinkResourceListener;
import org.apache.inlong.manager.service.listener.sort.StreamSortConfigListener;
import org.apache.inlong.manager.workflow.WorkflowContext;
import org.apache.inlong.manager.workflow.definition.ServiceTaskType;
import org.apache.inlong.manager.workflow.definition.TaskListenerFactory;
import org.apache.inlong.manager.workflow.event.task.QueueOperateListener;
+import org.apache.inlong.manager.workflow.event.task.ScheduleOperateListener;
import org.apache.inlong.manager.workflow.event.task.SinkOperateListener;
import org.apache.inlong.manager.workflow.event.task.SortOperateListener;
import org.apache.inlong.manager.workflow.event.task.SourceOperateListener;
@@ -53,6 +55,7 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
private List<QueueOperateListener> queueOperateListeners;
private List<SortOperateListener> sortOperateListeners;
private List<SinkOperateListener> sinkOperateListeners;
+ private List<ScheduleOperateListener> scheduleOperateListeners;
@Autowired
private StreamQueueResourceListener queueResourceListener;
@@ -60,6 +63,8 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
private StreamSortConfigListener streamSortConfigListener;
@Autowired
private StreamSinkResourceListener sinkResourceListener;
+ @Autowired
+ private StreamScheduleResourceListener scheduleResourceListener;
@PostConstruct
public void init() {
@@ -70,6 +75,8 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
sortOperateListeners.add(streamSortConfigListener);
sinkOperateListeners = new LinkedList<>();
sinkOperateListeners.add(sinkResourceListener);
+ scheduleOperateListeners = new LinkedList<>();
+ scheduleOperateListeners.add(scheduleResourceListener);
}
@Override
@@ -94,6 +101,10 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
if (CollectionUtils.isNotEmpty(pluginSinkOperateListeners)) {
sinkOperateListeners.addAll(pluginSinkOperateListeners);
}
+ List<ScheduleOperateListener> pluginScheduleOperateListeners = processPlugin.createScheduleOperateListeners();
+ if (CollectionUtils.isNotEmpty(pluginScheduleOperateListeners)) {
+ scheduleOperateListeners.addAll(pluginScheduleOperateListeners);
+ }
}
@Override
@@ -118,6 +129,9 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
case INIT_SINK:
List<SinkOperateListener> sinkOperateListeners = getSinkOperateListener(workflowContext);
return Lists.newArrayList(sinkOperateListeners);
+ case INIT_SCHEDULE:
+ List<ScheduleOperateListener> scheduleOperateListeners = getScheduleOperateListener(workflowContext);
+ return Lists.newArrayList(scheduleOperateListeners);
default:
throw new IllegalArgumentException(String.format("Unsupported ServiceTaskType %s", taskType));
}
@@ -131,6 +145,7 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
queueOperateListeners = new LinkedList<>();
sortOperateListeners = new LinkedList<>();
sinkOperateListeners = new LinkedList<>();
+ scheduleOperateListeners = new LinkedList<>();
}
/**
@@ -185,4 +200,17 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
return listeners;
}
+ /**
+ * Get schedule operate listener list.
+ */
+ private List<ScheduleOperateListener> getScheduleOperateListener(WorkflowContext context) {
+ List<ScheduleOperateListener> listeners = new ArrayList<>();
+ for (ScheduleOperateListener listener : scheduleOperateListeners) {
+ if (listener != null && listener.accept(context)) {
+ listeners.add(listener);
+ }
+ }
+ return listeners;
+ }
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/StreamScheduleResourceListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/StreamScheduleResourceListener.java
new file mode 100644
index 0000000000..61179dc343
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/StreamScheduleResourceListener.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.service.listener.schedule;
+
+import org.apache.inlong.manager.common.consts.InlongConstants;
+import org.apache.inlong.manager.common.enums.GroupOperateType;
+import org.apache.inlong.manager.common.enums.TaskEvent;
+import org.apache.inlong.manager.pojo.stream.InlongStreamExtInfo;
+import org.apache.inlong.manager.pojo.stream.InlongStreamInfo;
+import org.apache.inlong.manager.pojo.workflow.form.process.ProcessForm;
+import org.apache.inlong.manager.pojo.workflow.form.process.StreamResourceProcessForm;
+import org.apache.inlong.manager.workflow.WorkflowContext;
+import org.apache.inlong.manager.workflow.event.ListenerResult;
+import org.apache.inlong.manager.workflow.event.task.ScheduleOperateListener;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+@Service
+@Slf4j
+public class StreamScheduleResourceListener implements ScheduleOperateListener {
+
+ @Override
+ public TaskEvent event() {
+ return TaskEvent.COMPLETE;
+ }
+
+ @Override
+ public boolean accept(WorkflowContext context) {
+ ProcessForm processForm = context.getProcessForm();
+ String groupId = processForm.getInlongGroupId();
+ if (!(processForm instanceof StreamResourceProcessForm)) {
+ log.info("not add schedule stream listener, not StreamResourceProcessForm for groupId [{}]", groupId);
+ return false;
+ }
+
+ StreamResourceProcessForm streamProcessForm = (StreamResourceProcessForm) processForm;
+ String streamId = streamProcessForm.getStreamInfo().getInlongStreamId();
+ if (streamProcessForm.getGroupOperateType() != GroupOperateType.INIT) {
+ log.info("not add schedule stream listener, as the operate was not INIT for groupId [{}] streamId [{}]",
+ groupId, streamId);
+ return false;
+ }
+
+ log.info("add startup stream listener for groupId [{}] streamId [{}]", groupId, streamId);
+ return InlongConstants.DATASYNC_OFFLINE_MODE.equals(streamProcessForm.getGroupInfo().getInlongGroupMode());
+ }
+
+ @Override
+ public ListenerResult listen(WorkflowContext context) throws Exception {
+ StreamResourceProcessForm form = (StreamResourceProcessForm) context.getProcessForm();
+ InlongStreamInfo streamInfo = form.getStreamInfo();
+ final String groupId = streamInfo.getInlongGroupId();
+ final String streamId = streamInfo.getInlongStreamId();
+ log.info("begin to register schedule info for groupId={}, streamId={}", groupId, streamId);
+
+ // todo: register schedule info to schedule service
+
+ // after register schedule info successfully, add ext property to stream info
+ saveInfo(streamInfo, InlongConstants.REGISTER_SCHEDULE_STATUS,
+ InlongConstants.REGISTERED, streamInfo.getExtList());
+ log.info("success to register schedule info for group [" + groupId + "] and stream [" + streamId + "]");
+ return ListenerResult.success();
+ }
+
+ /**
+ * Save stream ext info into list.
+ */
+ private void saveInfo(InlongStreamInfo streamInfo, String keyName, String keyValue,
+ List<InlongStreamExtInfo> extInfoList) {
+ InlongStreamExtInfo extInfo = new InlongStreamExtInfo();
+ extInfo.setInlongGroupId(streamInfo.getInlongGroupId());
+ extInfo.setInlongStreamId(streamInfo.getInlongStreamId());
+ extInfo.setKeyName(keyName);
+ extInfo.setKeyValue(keyValue);
+ extInfoList.add(extInfo);
+ }
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java
index ebed9e72e3..f8981bf09a 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java
@@ -24,6 +24,7 @@ import org.apache.inlong.manager.common.enums.TaskEvent;
import org.apache.inlong.manager.common.exceptions.WorkflowListenerException;
import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.pojo.sink.StreamSink;
+import org.apache.inlong.manager.pojo.sort.util.StreamParseUtils;
import org.apache.inlong.manager.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.pojo.workflow.form.process.GroupResourceProcessForm;
import org.apache.inlong.manager.pojo.workflow.form.process.ProcessForm;
@@ -125,6 +126,13 @@ public class SortConfigListener implements SortOperateListener {
try {
for (InlongStreamInfo streamInfo : streamInfos) {
+ // do not build sort config if the group mode is offline and the stream is not config successfully
+ if (InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())
+ && !StreamParseUtils.isRegisterScheduleSuccess(streamInfo)) {
+ LOGGER.info("no need to build sort config for groupId={} streamId={} as the mode is offline "
+ + "and the stream is not config successfully yet", groupId, streamInfo.getInlongStreamId());
+ continue;
+ }
List<StreamSink> sinkList = streamInfo.getSinkList();
if (CollectionUtils.isEmpty(sinkList)) {
continue;
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java
index d3cf199a21..5505ed0694 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java
@@ -102,17 +102,26 @@ public class CreateStreamWorkflowDefinition implements WorkflowDefinition {
initSourceTask.setListenerFactory(streamTaskListenerFactory);
process.addTask(initSourceTask);
+ // Init Schedule info
+ ServiceTask initScheduleTask = new ServiceTask();
+ initScheduleTask.setName("InitSchedule");
+ initScheduleTask.setDisplayName("Stream-InitSchedule");
+ initScheduleTask.setServiceTaskType(ServiceTaskType.INIT_SCHEDULE);
+ initScheduleTask.setListenerFactory(streamTaskListenerFactory);
+ process.addTask(initScheduleTask);
+
// End node
EndEvent endEvent = new EndEvent();
process.setEndEvent(endEvent);
- // Task dependency order: 1.MQ -> 2.Sink -> 3.Sort -> 4.Source
+ // Task dependency order: 1.MQ -> 2.Sink -> 3.Sort -> 4.Source -> 5.Schedule
// To ensure that after some tasks fail, data will not start to be collected by source or consumed by sort
startEvent.addNext(initMQTask);
initMQTask.addNext(initSinkTask);
initSinkTask.addNext(initSortTask);
initSortTask.addNext(initSourceTask);
- initSourceTask.addNext(endEvent);
+ initSourceTask.addNext(initScheduleTask);
+ initScheduleTask.addNext(endEvent);
return process;
}
diff --git a/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/definition/ServiceTaskType.java b/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/definition/ServiceTaskType.java
index 2686b4b60a..156863a676 100644
--- a/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/definition/ServiceTaskType.java
+++ b/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/definition/ServiceTaskType.java
@@ -25,6 +25,7 @@ public enum ServiceTaskType {
INIT_MQ,
INIT_SORT,
INIT_SINK,
+ INIT_SCHEDULE,
STOP_SOURCE,
STOP_SORT,
diff --git a/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/definition/ServiceTaskType.java b/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/event/task/ScheduleOperateListener.java
similarity index 55%
copy from inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/definition/ServiceTaskType.java
copy to inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/event/task/ScheduleOperateListener.java
index 2686b4b60a..c3e162cd26 100644
--- a/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/definition/ServiceTaskType.java
+++ b/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/event/task/ScheduleOperateListener.java
@@ -15,30 +15,24 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.workflow.definition;
+package org.apache.inlong.manager.workflow.event.task;
-import java.util.Locale;
+import org.apache.inlong.manager.common.enums.TaskEvent;
+import org.apache.inlong.manager.workflow.WorkflowContext;
+import org.apache.inlong.manager.workflow.event.ListenerResult;
-public enum ServiceTaskType {
+public interface ScheduleOperateListener extends TaskEventListener {
- INIT_SOURCE,
- INIT_MQ,
- INIT_SORT,
- INIT_SINK,
+ ScheduleOperateListener DEFAULT_SCHEDULE_OPERATE_LISTENER = new ScheduleOperateListener() {
- STOP_SOURCE,
- STOP_SORT,
-
- RESTART_SOURCE,
- RESTART_SORT,
-
- DELETE_SOURCE,
- DELETE_MQ,
- DELETE_SORT;
-
- @Override
- public String toString() {
- return name().toLowerCase(Locale.ROOT);
- }
+ @Override
+ public TaskEvent event() {
+ return TaskEvent.COMPLETE;
+ }
+ @Override
+ public ListenerResult listen(WorkflowContext context) {
+ return ListenerResult.success();
+ }
+ };
}
diff --git a/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/plugin/ProcessPlugin.java b/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/plugin/ProcessPlugin.java
index 7e15d52761..452fa2bfb8 100644
--- a/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/plugin/ProcessPlugin.java
+++ b/inlong-manager/manager-workflow/src/main/java/org/apache/inlong/manager/workflow/plugin/ProcessPlugin.java
@@ -19,6 +19,7 @@ package org.apache.inlong.manager.workflow.plugin;
import org.apache.inlong.manager.common.plugin.Plugin;
import org.apache.inlong.manager.workflow.event.task.QueueOperateListener;
+import org.apache.inlong.manager.workflow.event.task.ScheduleOperateListener;
import org.apache.inlong.manager.workflow.event.task.SinkOperateListener;
import org.apache.inlong.manager.workflow.event.task.SortOperateListener;
import org.apache.inlong.manager.workflow.event.task.SourceOperateListener;
@@ -38,6 +39,10 @@ public interface ProcessPlugin extends Plugin {
return null;
}
+ default List<ScheduleOperateListener> createScheduleOperateListeners() {
+ return null;
+ }
+
default List<QueueOperateListener> createQueueOperateListeners() {
return null;
}
(inlong) 05/09: [INLONG-10247][Manager] Support schedule information management for offline sync (#10254)
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 2afe763e0d32c9fed845943fe8c16d2a218559a0
Author: AloysZhang <al...@apache.org>
AuthorDate: Thu May 23 18:18:31 2024 +0800
[INLONG-10247][Manager] Support schedule information management for offline sync (#10254)
---
.../client/api/inner/client/ClientFactory.java | 2 +
.../api/inner/client/InLongScheduleClient.java | 67 +++++++++
.../client/api/service/InLongScheduleApi.java | 49 +++++++
.../inlong/manager/common/enums/ErrorCodeEnum.java | 3 +
.../manager/common/enums/OperationTarget.java | 4 +-
.../{OperationTarget.java => ScheduleStatus.java} | 54 ++++---
.../inlong/manager/dao/entity/ScheduleEntity.java | 61 ++++++++
.../manager/dao/mapper/ScheduleEntityMapper.java} | 36 ++---
.../resources/mappers/ScheduleEntityMapper.xml | 144 +++++++++++++++++++
.../manager/dao/mapper/ScheduleEntityTest.java | 118 ++++++++++++++++
.../inlong/manager/pojo/schedule/ScheduleInfo.java | 82 +++++++++++
.../manager/pojo/schedule/ScheduleInfoRequest.java | 76 ++++++++++
.../manager/service/schedule/ScheduleService.java | 71 ++++++++++
.../service/schedule/ScheduleServiceImpl.java | 155 +++++++++++++++++++++
.../main/resources/h2/apache_inlong_manager.sql | 31 +++++
.../manager-web/sql/apache_inlong_manager.sql | 28 ++++
inlong-manager/manager-web/sql/changes-1.13.0.sql | 30 ++++
.../web/controller/InLongSchedulerController.java | 88 ++++++++++++
18 files changed, 1043 insertions(+), 56 deletions(-)
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/ClientFactory.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/ClientFactory.java
index 3c67f1fa71..01c9fb6473 100644
--- a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/ClientFactory.java
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/ClientFactory.java
@@ -55,6 +55,7 @@ public class ClientFactory {
private final AuditClient auditClient;
private final InlongTenantClient inlongTenantClient;
private final InlongTenantRoleClient inlongTenantRoleClient;
+ private final InLongScheduleClient inLongScheduleClient;
public ClientFactory(ClientConfiguration configuration) {
groupClient = new InlongGroupClient(configuration);
@@ -74,5 +75,6 @@ public class ClientFactory {
auditClient = new AuditClient(configuration);
inlongTenantClient = new InlongTenantClient(configuration);
inlongTenantRoleClient = new InlongTenantRoleClient(configuration);
+ inLongScheduleClient = new InLongScheduleClient(configuration);
}
}
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/InLongScheduleClient.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/InLongScheduleClient.java
new file mode 100644
index 0000000000..86638dbae8
--- /dev/null
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/inner/client/InLongScheduleClient.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.client.api.inner.client;
+
+import org.apache.inlong.manager.client.api.ClientConfiguration;
+import org.apache.inlong.manager.client.api.service.InLongScheduleApi;
+import org.apache.inlong.manager.client.api.util.ClientUtils;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.pojo.common.Response;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
+
+public class InLongScheduleClient {
+
+ private InLongScheduleApi scheduleApi;
+
+ public InLongScheduleClient(ClientConfiguration clientConfiguration) {
+ scheduleApi = ClientUtils.createRetrofit(clientConfiguration).create(InLongScheduleApi.class);
+ }
+
+ public Integer createScheduleInfo(ScheduleInfoRequest request) {
+ Response<Integer> response = ClientUtils.executeHttpCall(scheduleApi.createSchedule(request));
+ ClientUtils.assertRespSuccess(response);
+ return response.getData();
+ }
+
+ public Boolean scheduleInfoExist(String groupId) {
+ Preconditions.expectNotBlank(groupId, ErrorCodeEnum.GROUP_ID_IS_EMPTY);
+ Response<Boolean> response = ClientUtils.executeHttpCall(scheduleApi.exist(groupId));
+ ClientUtils.assertRespSuccess(response);
+ return response.getData();
+ }
+
+ public Boolean updateScheduleInfo(ScheduleInfoRequest request) {
+ Response<Boolean> response = ClientUtils.executeHttpCall(scheduleApi.update(request));
+ ClientUtils.assertRespSuccess(response);
+ return response.getData();
+ }
+
+ public ScheduleInfo getScheduleInfo(String groupId) {
+ Response<ScheduleInfo> response = ClientUtils.executeHttpCall(scheduleApi.get(groupId));
+ ClientUtils.assertRespSuccess(response);
+ return response.getData();
+ }
+
+ public Boolean deleteScheduleInfo(String groupId) {
+ Response<Boolean> response = ClientUtils.executeHttpCall(scheduleApi.delete(groupId));
+ ClientUtils.assertRespSuccess(response);
+ return response.getData();
+ }
+}
diff --git a/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/InLongScheduleApi.java b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/InLongScheduleApi.java
new file mode 100644
index 0000000000..0da7be3d6f
--- /dev/null
+++ b/inlong-manager/manager-client/src/main/java/org/apache/inlong/manager/client/api/service/InLongScheduleApi.java
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.client.api.service;
+
+import org.apache.inlong.manager.pojo.common.Response;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
+
+import retrofit2.Call;
+import retrofit2.http.Body;
+import retrofit2.http.DELETE;
+import retrofit2.http.GET;
+import retrofit2.http.POST;
+import retrofit2.http.Path;
+import retrofit2.http.Query;
+
+public interface InLongScheduleApi {
+
+ @POST("schedule/save")
+ Call<Response<Integer>> createSchedule(@Body ScheduleInfoRequest request);
+
+ @GET("schedule/exist/{groupId}")
+ Call<Response<Boolean>> exist(@Path("groupId") String groupId);
+
+ @POST("schedule/update")
+ Call<Response<Boolean>> update(@Body ScheduleInfoRequest request);
+
+ @GET("schedule/get")
+ Call<Response<ScheduleInfo>> get(@Query("groupId") String groupId);
+
+ @DELETE("schedule/delete/{groupId}")
+ Call<Response<Boolean>> delete(@Path("groupId") String groupId);
+
+}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java
index ed7549b699..6a8f9b4699 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java
@@ -125,6 +125,9 @@ public enum ErrorCodeEnum {
MQ_TYPE_IS_NULL(1600, "MQ type is null"),
MQ_TYPE_NOT_SUPPORT(1601, "MQ type '%s' not support"),
+ SCHEDULE_NOT_FOUND(1700, "Schedule info not found"),
+ SCHEDULE_DUPLICATE(1701, "Schedule info already exist"),
+
WORKFLOW_EXE_FAILED(4000, "Workflow execution exception"),
WORKFLOW_APPROVER_NOT_FOUND(4001, "Workflow approver does not exist/no operation authority"),
WORKFLOW_DELETE_RECORD_FAILED(4002, "Workflow delete record failure"),
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java
index d92fb31b33..cb9e1638be 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java
@@ -46,6 +46,8 @@ public enum OperationTarget {
TENANT_ROLE,
- TEMPLATE
+ TEMPLATE,
+
+ SCHEDULE
}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java
similarity index 58%
copy from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java
copy to inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java
index d92fb31b33..cb256a491c 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java
@@ -17,35 +17,29 @@
package org.apache.inlong.manager.common.enums;
-/**
- * Operation target
- */
-public enum OperationTarget {
-
- TENANT,
-
- GROUP,
-
- STREAM,
-
- SOURCE,
-
- SINK,
-
- CONSUME,
-
- WORKFLOW,
-
- NODE,
-
- CLUSTER,
-
- TRANSFORM,
-
- INLONG_ROLE,
-
- TENANT_ROLE,
-
- TEMPLATE
+import lombok.Getter;
+
+@Getter
+public enum ScheduleStatus {
+
+ NEW(100, "new"),
+ DELETED(40, "deleted");
+
+ private final Integer code;
+ private final String description;
+
+ ScheduleStatus(Integer code, String description) {
+ this.code = code;
+ this.description = description;
+ }
+
+ public static ScheduleStatus forCode(int code) {
+ for (ScheduleStatus status : values()) {
+ if (status.getCode() == code) {
+ return status;
+ }
+ }
+ throw new IllegalStateException(String.format("Illegal code=%s for ScheduleStatus", code));
+ }
}
diff --git a/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/ScheduleEntity.java b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/ScheduleEntity.java
new file mode 100644
index 0000000000..75237343cb
--- /dev/null
+++ b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/entity/ScheduleEntity.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.dao.entity;
+
+import lombok.Data;
+
+import java.io.Serializable;
+import java.sql.Timestamp;
+import java.util.Date;
+
+@Data
+public class ScheduleEntity implements Serializable {
+
+ private static final long serialVersionUID = 1L;
+
+ private Integer id;
+ // inLong group id
+ private String inlongGroupId;
+ // schedule type, support [normal, crontab], 0 for normal and 1 for crontab
+ private Integer scheduleType;
+ // time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
+ // M=month, W=week, D=day, H=hour, M=minute, O=oneway
+ private String scheduleUnit;
+ private Integer scheduleInterval;
+ // schedule start time, long type timestamp
+ private Timestamp startTime;
+ // schedule end time, long type timestamp
+ private Timestamp endTime;
+ // delay time to start task, in minutes
+ private Integer delayTime;
+ // if task depend on itself
+ private Integer selfDepend;
+ private Integer taskParallelism;
+ // expression of crontab, used when scheduleType is crontab
+ private String crontabExpression;
+
+ private Integer status;
+ private Integer previousStatus;
+ private Integer isDeleted;
+ private String creator;
+ private String modifier;
+ private Date createTime;
+ private Date modifyTime;
+ private Integer version;
+
+}
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/ScheduleEntityMapper.java
similarity index 60%
copy from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java
copy to inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/ScheduleEntityMapper.java
index d92fb31b33..23e8e23c31 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/OperationTarget.java
+++ b/inlong-manager/manager-dao/src/main/java/org/apache/inlong/manager/dao/mapper/ScheduleEntityMapper.java
@@ -15,37 +15,23 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.common.enums;
+package org.apache.inlong.manager.dao.mapper;
-/**
- * Operation target
- */
-public enum OperationTarget {
-
- TENANT,
-
- GROUP,
-
- STREAM,
-
- SOURCE,
-
- SINK,
-
- CONSUME,
-
- WORKFLOW,
+import org.apache.inlong.manager.dao.entity.ScheduleEntity;
- NODE,
+import org.apache.ibatis.annotations.Param;
+import org.springframework.stereotype.Repository;
- CLUSTER,
+@Repository
+public interface ScheduleEntityMapper {
- TRANSFORM,
+ int insert(ScheduleEntity scheduleEntity);
- INLONG_ROLE,
+ ScheduleEntity selectByPrimaryKey(Integer id);
- TENANT_ROLE,
+ ScheduleEntity selectByGroupId(String groupId);
- TEMPLATE
+ int updateByIdSelective(ScheduleEntity scheduleEntity);
+ int deleteByGroupId(@Param("inlongGroupId") String inlongGroupId);
}
diff --git a/inlong-manager/manager-dao/src/main/resources/mappers/ScheduleEntityMapper.xml b/inlong-manager/manager-dao/src/main/resources/mappers/ScheduleEntityMapper.xml
new file mode 100644
index 0000000000..d719aa8988
--- /dev/null
+++ b/inlong-manager/manager-dao/src/main/resources/mappers/ScheduleEntityMapper.xml
@@ -0,0 +1,144 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you 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
+ http://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.
+-->
+
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="org.apache.inlong.manager.dao.mapper.ScheduleEntityMapper">
+ <resultMap id="BaseResultMap" type="org.apache.inlong.manager.dao.entity.ScheduleEntity">
+ <id column="id" jdbcType="INTEGER" property="id"/>
+ <result column="inlong_group_id" jdbcType="VARCHAR" property="inlongGroupId"/>
+ <result column="schedule_type" jdbcType="INTEGER" property="scheduleType"/>
+ <result column="schedule_unit" jdbcType="VARCHAR" property="scheduleUnit"/>
+ <result column="schedule_interval" jdbcType="INTEGER" property="scheduleInterval"/>
+ <result column="start_time" jdbcType="TIMESTAMP" property="startTime"/>
+ <result column="end_time" jdbcType="TIMESTAMP" property="endTime"/>
+ <result column="delay_time" jdbcType="INTEGER" property="delayTime"/>
+ <result column="self_depend" jdbcType="INTEGER" property="selfDepend"/>
+ <result column="task_parallelism" jdbcType="INTEGER" property="taskParallelism"/>
+ <result column="crontab_expression" jdbcType="VARCHAR" property="crontabExpression"/>
+
+ <result column="status" jdbcType="INTEGER" property="status"/>
+ <result column="previous_status" jdbcType="INTEGER" property="previousStatus"/>
+ <result column="is_deleted" jdbcType="INTEGER" property="isDeleted"/>
+ <result column="creator" jdbcType="VARCHAR" property="creator"/>
+ <result column="modifier" jdbcType="VARCHAR" property="modifier"/>
+ <result column="create_time" jdbcType="TIMESTAMP" property="createTime"/>
+ <result column="modify_time" jdbcType="TIMESTAMP" property="modifyTime"/>
+ <result column="version" jdbcType="INTEGER" property="version"/>
+ </resultMap>
+
+ <sql id="Base_Column_List">
+ id, inlong_group_id, schedule_type, schedule_unit, schedule_interval, start_time,
+ end_time, delay_time, self_depend, task_parallelism, crontab_expression,
+ status, previous_status, is_deleted, creator, modifier, create_time, modify_time, version
+ </sql>
+
+ <insert id="insert" useGeneratedKeys="true" keyProperty="id"
+ parameterType="org.apache.inlong.manager.dao.entity.ScheduleEntity">
+ insert into schedule_config (id, inlong_group_id, schedule_type, schedule_unit,
+ schedule_interval, start_time, end_time, delay_time,
+ self_depend, task_parallelism, crontab_expression,
+ status, previous_status, creator, modifier)
+ values (#{id, jdbcType=INTEGER}, #{inlongGroupId, jdbcType=VARCHAR},
+ #{scheduleType, jdbcType=INTEGER}, #{scheduleUnit, jdbcType=VARCHAR},
+ #{scheduleInterval, jdbcType=INTEGER}, #{startTime, jdbcType=TIMESTAMP},
+ #{endTime, jdbcType=TIMESTAMP}, #{delayTime, jdbcType=INTEGER},
+ #{selfDepend, jdbcType=INTEGER}, #{taskParallelism, jdbcType=INTEGER},
+ #{crontabExpression, jdbcType=VARCHAR}, #{status,jdbcType=INTEGER},
+ #{previousStatus,jdbcType=INTEGER}, #{creator,jdbcType=VARCHAR},
+ #{modifier,jdbcType=VARCHAR})
+ </insert>
+
+ <select id="selectByPrimaryKey" parameterType="java.lang.Integer" resultMap="BaseResultMap">
+ select
+ <include refid="Base_Column_List"/>
+ from schedule_config
+ where id = #{id,jdbcType=INTEGER}
+ and is_deleted = 0
+ </select>
+
+ <select id="selectByGroupId" parameterType="java.lang.String" resultMap="BaseResultMap">
+ select
+ <include refid="Base_Column_List"/>
+ from schedule_config
+ where inlong_group_id = #{inlongGroupId,jdbcType=VARCHAR}
+ and is_deleted = 0
+ </select>
+
+ <update id="updateByIdSelective" parameterType="org.apache.inlong.manager.dao.entity.ScheduleEntity">
+ update schedule_config
+ <set>
+ <if test="inlongGroupId != null">
+ inlong_group_id = #{inlongGroupId, jdbcType=VARCHAR},
+ </if>
+ <if test="scheduleType != null">
+ schedule_type = #{scheduleType, jdbcType=INTEGER},
+ </if>
+ <if test="scheduleUnit !=null">
+ schedule_unit = #{scheduleUnit, jdbcType=VARCHAR},
+ </if>
+ <if test="scheduleInterval != null">
+ schedule_interval = #{scheduleInterval, jdbcType=INTEGER},
+ </if>
+ <if test="startTime != null">
+ start_time = #{startTime, jdbcType=TIMESTAMP},
+ </if>
+ <if test="endTime != null">
+ end_time = #{endTime, jdbcType=TIMESTAMP},
+ </if>
+ <if test="delayTime != null">
+ delay_time = #{delayTime, jdbcType=INTEGER},
+ </if>
+ <if test="selfDepend != null">
+ self_depend = #{selfDepend, jdbcType=INTEGER},
+ </if>
+ <if test="taskParallelism != null">
+ task_parallelism = #{taskParallelism, jdbcType=INTEGER},
+ </if>
+ <if test="crontabExpression != null">
+ crontab_expression = #{crontabExpression, jdbcType=VARCHAR},
+ </if>
+ <if test="status != null">
+ status = #{status, jdbcType=VARCHAR},
+ </if>
+ <if test="previousStatus != null">
+ previous_status = #{previousStatus, jdbcType=VARCHAR},
+ </if>
+ <if test="isDeleted != null">
+ is_deleted = #{isDeleted},
+ </if>
+ <if test="creator != null">
+ creator = #{creator, jdbcType=VARCHAR},
+ </if>
+ <if test="modifier != null">
+ modifier = #{modifier, jdbcType=VARCHAR},
+ </if>
+ version = #{version, jdbcType=INTEGER} + 1
+ </set>
+ <where>
+ id = #{id, jdbcType=INTEGER}
+ and version = #{version,jdbcType=INTEGER}
+ </where>
+ </update>
+
+ <delete id="deleteByGroupId">
+ delete
+ from schedule_config
+ where inlong_group_id = #{inlongGroupId, jdbcType=VARCHAR}
+ </delete>
+
+</mapper>
\ No newline at end of file
diff --git a/inlong-manager/manager-dao/src/test/java/org/apache/inlong/manager/dao/mapper/ScheduleEntityTest.java b/inlong-manager/manager-dao/src/test/java/org/apache/inlong/manager/dao/mapper/ScheduleEntityTest.java
new file mode 100644
index 0000000000..ef4207bc84
--- /dev/null
+++ b/inlong-manager/manager-dao/src/test/java/org/apache/inlong/manager/dao/mapper/ScheduleEntityTest.java
@@ -0,0 +1,118 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.dao.mapper;
+
+import org.apache.inlong.manager.dao.DaoBaseTest;
+import org.apache.inlong.manager.dao.entity.ScheduleEntity;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import java.sql.Timestamp;
+import java.util.Date;
+
+public class ScheduleEntityTest extends DaoBaseTest {
+
+ public static final String GROUP_ID_PREFIX = "test_group_";
+ public static final String USER = "admin";
+ public static final int SCHEDULE_TYPE = 0;
+ public static final int SCHEDULE_TYPE_NEW = 1;
+ public static final String SCHEDULE_UNIT = "H";
+ public static final String SCHEDULE_UNIT_NEW = "D";
+ public static final int SCHEDULE_INTERVAL = 1;
+ public static final int SCHEDULE_INTERVAL_NEW = 1;
+ public static final Timestamp DEFAULT_TIME = new Timestamp(System.currentTimeMillis());
+
+ @Autowired
+ ScheduleEntityMapper scheduleEntityMapper;
+
+ @Test
+ public void testSelectByGroupId() throws Exception {
+ ScheduleEntity scheduleEntity = genEntity();
+ scheduleEntityMapper.insert(scheduleEntity);
+ ScheduleEntity entityQueried = scheduleEntityMapper.selectByGroupId(scheduleEntity.getInlongGroupId());
+ Assertions.assertEquals(scheduleEntity.getInlongGroupId(), entityQueried.getInlongGroupId());
+ Assertions.assertEquals(SCHEDULE_TYPE, entityQueried.getScheduleType());
+ Assertions.assertEquals(SCHEDULE_UNIT, entityQueried.getScheduleUnit());
+ Assertions.assertEquals(SCHEDULE_INTERVAL, entityQueried.getScheduleInterval());
+ Assertions.assertEquals(DEFAULT_TIME, entityQueried.getStartTime());
+ Assertions.assertEquals(DEFAULT_TIME, entityQueried.getEndTime());
+ Assertions.assertEquals(USER, entityQueried.getCreator());
+ }
+
+ @Test
+ public void testUpdate() throws Exception {
+ ScheduleEntity scheduleEntity = genEntity();
+ scheduleEntityMapper.insert(scheduleEntity);
+ ScheduleEntity entityQueried = scheduleEntityMapper.selectByGroupId(scheduleEntity.getInlongGroupId());
+ Assertions.assertEquals(scheduleEntity.getInlongGroupId(), entityQueried.getInlongGroupId());
+ Assertions.assertEquals(SCHEDULE_TYPE, entityQueried.getScheduleType());
+ Assertions.assertEquals(SCHEDULE_UNIT, entityQueried.getScheduleUnit());
+ Assertions.assertEquals(SCHEDULE_INTERVAL, entityQueried.getScheduleInterval());
+ Assertions.assertEquals(DEFAULT_TIME, entityQueried.getStartTime());
+ Assertions.assertEquals(DEFAULT_TIME, entityQueried.getEndTime());
+ Assertions.assertEquals(USER, entityQueried.getCreator());
+
+ entityQueried.setScheduleType(SCHEDULE_TYPE_NEW);
+ entityQueried.setScheduleUnit(SCHEDULE_UNIT_NEW);
+ entityQueried.setScheduleInterval(SCHEDULE_INTERVAL_NEW);
+ scheduleEntityMapper.updateByIdSelective(entityQueried);
+ entityQueried = scheduleEntityMapper.selectByGroupId(scheduleEntity.getInlongGroupId());
+ Assertions.assertEquals(scheduleEntity.getInlongGroupId(), entityQueried.getInlongGroupId());
+ Assertions.assertEquals(SCHEDULE_TYPE_NEW, entityQueried.getScheduleType());
+ Assertions.assertEquals(SCHEDULE_UNIT_NEW, entityQueried.getScheduleUnit());
+ Assertions.assertEquals(SCHEDULE_INTERVAL_NEW, entityQueried.getScheduleInterval());
+ Assertions.assertEquals(DEFAULT_TIME, entityQueried.getStartTime());
+ Assertions.assertEquals(DEFAULT_TIME, entityQueried.getEndTime());
+ Assertions.assertEquals(USER, entityQueried.getCreator());
+ }
+
+ @Test
+ public void testDelete() throws Exception {
+ ScheduleEntity scheduleEntity = genEntity();
+ scheduleEntityMapper.insert(scheduleEntity);
+ ScheduleEntity entityQueried = scheduleEntityMapper.selectByGroupId(scheduleEntity.getInlongGroupId());
+ Assertions.assertEquals(scheduleEntity.getInlongGroupId(), entityQueried.getInlongGroupId());
+ Assertions.assertEquals(SCHEDULE_TYPE, entityQueried.getScheduleType());
+ Assertions.assertEquals(SCHEDULE_UNIT, entityQueried.getScheduleUnit());
+ Assertions.assertEquals(SCHEDULE_INTERVAL, entityQueried.getScheduleInterval());
+ Assertions.assertEquals(DEFAULT_TIME, entityQueried.getStartTime());
+ Assertions.assertEquals(DEFAULT_TIME, entityQueried.getEndTime());
+ Assertions.assertEquals(USER, entityQueried.getCreator());
+
+ scheduleEntityMapper.deleteByGroupId(scheduleEntity.getInlongGroupId());
+ entityQueried = scheduleEntityMapper.selectByGroupId(scheduleEntity.getInlongGroupId());
+ Assertions.assertNull(entityQueried);
+ }
+
+ private ScheduleEntity genEntity() {
+ ScheduleEntity entity = new ScheduleEntity();
+ entity.setInlongGroupId(GROUP_ID_PREFIX + System.currentTimeMillis());
+ entity.setScheduleType(SCHEDULE_TYPE);
+ entity.setScheduleUnit(SCHEDULE_UNIT);
+ entity.setScheduleInterval(SCHEDULE_INTERVAL);
+ entity.setStartTime(DEFAULT_TIME);
+ entity.setEndTime(DEFAULT_TIME);
+ entity.setCreator(USER);
+ entity.setCreateTime(new Date());
+ entity.setModifyTime(new Date());
+ entity.setIsDeleted(0);
+ return entity;
+ }
+}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java
new file mode 100644
index 0000000000..13afce70a4
--- /dev/null
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java
@@ -0,0 +1,82 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.pojo.schedule;
+
+import org.apache.inlong.manager.common.validation.UpdateValidation;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import javax.validation.constraints.NotNull;
+
+import java.sql.Timestamp;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+@ApiModel("Schedule response")
+public class ScheduleInfo {
+
+ @ApiModelProperty(value = "Primary key")
+ @NotNull(groups = UpdateValidation.class)
+ private Integer id;
+
+ @ApiModelProperty("Inlong Group ID")
+ @NotNull
+ private String inlongGroupId;
+
+ // schedule type, support [normal, crontab], 0 for normal and 1 for crontab
+ @ApiModelProperty("Schedule type")
+ private Integer scheduleType;
+
+ // time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
+ // M=month, W=week, D=day, H=hour, M=minute, O=oneway
+ @ApiModelProperty("TimeUnit for schedule interval")
+ private String scheduleUnit;
+
+ @ApiModelProperty("Schedule interval")
+ private Integer scheduleInterval;
+
+ @ApiModelProperty("Start time")
+ private Timestamp startTime;
+
+ @ApiModelProperty("End time")
+ private Timestamp endTime;
+
+ @ApiModelProperty("Delay time")
+ private Integer delayTime;
+
+ @ApiModelProperty("Self depend")
+ private Integer selfDepend;
+
+ @ApiModelProperty("Schedule task parallelism")
+ private Integer taskParallelism;
+
+ @ApiModelProperty("Schedule task parallelism")
+ private Integer crontabExpression;
+
+ @ApiModelProperty(value = "Version number")
+ @NotNull(groups = UpdateValidation.class, message = "version cannot be null")
+ private Integer version;
+
+}
\ No newline at end of file
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java
new file mode 100644
index 0000000000..eff1660719
--- /dev/null
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.pojo.schedule;
+
+import org.apache.inlong.manager.common.validation.UpdateValidation;
+
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+import javax.validation.constraints.NotNull;
+
+import java.sql.Timestamp;
+
+@Data
+@ApiModel("Schedule request")
+public class ScheduleInfoRequest {
+
+ @ApiModelProperty(value = "Primary key")
+ @NotNull(groups = UpdateValidation.class)
+ private Integer id;
+
+ @ApiModelProperty("Inlong Group ID")
+ @NotNull
+ private String inlongGroupId;
+
+ // schedule type, support [normal, crontab], 0 for normal and 1 for crontab
+ @ApiModelProperty("Schedule type")
+ private Integer scheduleType;
+
+ // time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
+ // M=month, W=week, D=day, H=hour, M=minute, O=oneway
+ @ApiModelProperty("TimeUnit for schedule interval")
+ private String scheduleUnit;
+
+ @ApiModelProperty("Schedule interval")
+ private Integer scheduleInterval;
+
+ @ApiModelProperty("Start time")
+ private Timestamp startTime;
+
+ @ApiModelProperty("End time")
+ private Timestamp endTime;
+
+ @ApiModelProperty("Delay time")
+ private Integer delayTime;
+
+ @ApiModelProperty("Self depend")
+ private Integer selfDepend;
+
+ @ApiModelProperty("Schedule task parallelism")
+ private Integer taskParallelism;
+
+ @ApiModelProperty("Schedule task parallelism")
+ private Integer crontabExpression;
+
+ @ApiModelProperty(value = "Version number")
+ @NotNull(groups = UpdateValidation.class, message = "version cannot be null")
+ private Integer version;
+
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleService.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleService.java
new file mode 100644
index 0000000000..d00e7134d4
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleService.java
@@ -0,0 +1,71 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.service.schedule;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
+
+import javax.validation.Valid;
+import javax.validation.constraints.NotNull;
+
+public interface ScheduleService {
+
+ /**
+ * Save schedule info.
+ *
+ * @param request schedule request need to save
+ * @param operator name of operator
+ * @return schedule info id in backend storage
+ */
+ int save(@Valid @NotNull(message = "schedule request cannot be null") ScheduleInfoRequest request,
+ String operator);
+
+ /**
+ * Query whether schedule info exists for specified inlong group
+ *
+ * @param groupId the group id to be queried
+ * @return does it exist
+ */
+ Boolean exist(String groupId);
+
+ /**
+ * Get schedule info based on inlong group id
+ *
+ * @param groupId inlong group id
+ * @return detail of inlong group
+ */
+ ScheduleInfo get(String groupId);
+
+ /**
+ * Modify schedule information
+ *
+ * @param request schedule request that needs to be modified
+ * @param operator name of operator
+ * @return whether succeed
+ */
+ Boolean update(@Valid @NotNull(message = "schedule request cannot be null") ScheduleInfoRequest request,
+ String operator);
+
+ /**
+ * Delete schedule info for gropuId.
+ * @param groupId groupId to find a schedule info to delete
+ * @param operator name of operator
+ * @Return whether succeed
+ * */
+ Boolean deleteByGroupId(String groupId, String operator);
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleServiceImpl.java
new file mode 100644
index 0000000000..480189da9e
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleServiceImpl.java
@@ -0,0 +1,155 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.service.schedule;
+
+import org.apache.inlong.manager.common.consts.InlongConstants;
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.enums.ScheduleStatus;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+import org.apache.inlong.manager.common.util.CommonBeanUtils;
+import org.apache.inlong.manager.common.util.Preconditions;
+import org.apache.inlong.manager.dao.entity.InlongGroupEntity;
+import org.apache.inlong.manager.dao.entity.ScheduleEntity;
+import org.apache.inlong.manager.dao.mapper.InlongGroupEntityMapper;
+import org.apache.inlong.manager.dao.mapper.ScheduleEntityMapper;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+import java.util.Objects;
+
+@Service
+public class ScheduleServiceImpl implements ScheduleService {
+
+ private static Logger LOGGER = LoggerFactory.getLogger(ScheduleServiceImpl.class);
+
+ @Autowired
+ private InlongGroupEntityMapper groupMapper;
+ @Autowired
+ private ScheduleEntityMapper scheduleEntityMapper;
+
+ @Override
+ public int save(ScheduleInfoRequest request, String operator) {
+ LOGGER.debug("begin to save schedule info, scheduleInfo={}, operator={}", request, operator);
+
+ String groupId = request.getInlongGroupId();
+ checkGroupExist(groupId);
+ if (scheduleEntityMapper.selectByGroupId(groupId) != null) {
+ LOGGER.error("schedule info for group={} already exists", groupId);
+ throw new BusinessException(ErrorCodeEnum.SCHEDULE_DUPLICATE);
+ }
+
+ ScheduleEntity scheduleEntity = CommonBeanUtils.copyProperties(request, ScheduleEntity::new);
+ scheduleEntity.setStatus(ScheduleStatus.NEW.getCode());
+ scheduleEntity.setCreator(operator);
+ scheduleEntity.setModifier(operator);
+ return scheduleEntityMapper.insert(scheduleEntity);
+ }
+
+ @Override
+ public Boolean exist(String groupId) {
+ checkGroupExist(groupId);
+ return scheduleEntityMapper.selectByGroupId(groupId) != null;
+ }
+
+ @Override
+ public ScheduleInfo get(String groupId) {
+ LOGGER.debug("begin to get schedule info by groupId={}", groupId);
+ ScheduleEntity entity = getScheduleEntity(groupId);
+ return CommonBeanUtils.copyProperties(entity, ScheduleInfo::new);
+ }
+
+ @Override
+ public Boolean update(ScheduleInfoRequest request, String operator) {
+ LOGGER.debug("begin to update schedule info={}", request);
+ String groupId = request.getInlongGroupId();
+ ScheduleEntity entity = getScheduleEntity(groupId);
+ String errMsg =
+ String.format("schedule info has already been updated with groupId=%s, curVersion=%s, expectVersion=%s",
+ entity.getInlongGroupId(), request.getVersion(), entity.getVersion());
+ if (!Objects.equals(entity.getVersion(), request.getVersion())) {
+ LOGGER.error(errMsg);
+ throw new BusinessException(ErrorCodeEnum.CONFIG_EXPIRED);
+ }
+ CommonBeanUtils.copyProperties(request, entity, true);
+ entity.setModifier(operator);
+ updateScheduleInfo(entity, errMsg);
+ LOGGER.info("success to update schedule info for groupId={}", groupId);
+ return true;
+ }
+
+ @Override
+ public Boolean deleteByGroupId(String groupId, String operator) {
+ LOGGER.debug("begin to delete schedule info for groupId={}", groupId);
+ checkGroupExist(groupId);
+ ScheduleEntity entity = scheduleEntityMapper.selectByGroupId(groupId);
+ if (entity == null) {
+ LOGGER.error("schedule info for groupId={} does not exist", groupId);
+ return false;
+ }
+ entity.setPreviousStatus(entity.getStatus());
+ entity.setStatus(ScheduleStatus.DELETED.getCode());
+ entity.setModifier(operator);
+ entity.setIsDeleted(entity.getId());
+ updateScheduleInfo(entity,
+ String.format("schedule info has already been updated with groupId=%s, curVersion=%s",
+ entity.getInlongGroupId(), entity.getVersion()));
+ LOGGER.info("success to delete schedule info for groupId={}", groupId);
+ return true;
+ }
+
+ /**
+ * Check whether InLongGroup exists, throw BusinessException with ErrorCodeEnum.GROUP_NOT_FOUND if check failed.
+ * */
+ private void checkGroupExist(String groupId) {
+ Preconditions.expectNotBlank(groupId, ErrorCodeEnum.GROUP_ID_IS_EMPTY);
+ InlongGroupEntity entity = groupMapper.selectByGroupId(groupId);
+ if (entity == null) {
+ LOGGER.error("inlong group not found by groupId={}", groupId);
+ throw new BusinessException(ErrorCodeEnum.GROUP_NOT_FOUND);
+ }
+ }
+
+ private ScheduleEntity getScheduleEntity(String groupId) {
+ checkGroupExist(groupId);
+ ScheduleEntity entity = scheduleEntityMapper.selectByGroupId(groupId);
+ if (entity == null) {
+ LOGGER.error("schedule info for group={} not found", groupId);
+ throw new BusinessException(ErrorCodeEnum.SCHEDULE_NOT_FOUND);
+ }
+ return entity;
+ }
+
+ /**
+ * Update schedule entity and throw exception if update failed.
+ * @param entity to update
+ * @param errorMsg when update failed.
+ * @return
+ *
+ * */
+ private void updateScheduleInfo(ScheduleEntity entity, String errorMsg) {
+ if (scheduleEntityMapper.updateByIdSelective(entity) != InlongConstants.AFFECTED_ONE_ROW) {
+ LOGGER.error(errorMsg);
+ throw new BusinessException(ErrorCodeEnum.CONFIG_EXPIRED);
+ }
+ }
+}
diff --git a/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql b/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql
index 3864ffcbe3..a8b545f7bf 100644
--- a/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql
+++ b/inlong-manager/manager-test/src/main/resources/h2/apache_inlong_manager.sql
@@ -962,4 +962,35 @@ CREATE TABLE IF NOT EXISTS `tenant_template`
-- ----------------------------
+-- ----------------------------
+-- Table structure for schedule_config
+-- ----------------------------
+CREATE TABLE IF NOT EXISTS `schedule_config`
+(
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
+ `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, undeleted ones cannot be repeated',
+ `schedule_type` int(4) NOT NULL DEFAULT '0' COMMENT 'Schedule type, 0 for normal, 1 for crontab',
+ `schedule_unit` varchar(64) NOT NULL COMMENT 'Schedule unit,M=month, W=week, D=day, H=hour, M=minute, O=oneway',
+ `schedule_interval` int(11) DEFAULT '1' COMMENT 'Schedule interval',
+ `start_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Start time for schedule',
+ `end_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'End time for schedule',
+ `delay_time` int(11) DEFAULT '0' COMMENT 'Delay time in minutes to schedule',
+ `self_depend` int(11) DEFAULT NULL COMMENT 'Self depend info',
+ `task_parallelism` int(11) DEFAULT NULL COMMENT 'Task parallelism',
+ `crontab_expression` varchar(256) DEFAULT NULL COMMENT 'Crontab expression if schedule type is crontab',
+ `status` int(4) DEFAULT '100' COMMENT 'Schedule status',
+ `previous_status` int(4) DEFAULT '100' COMMENT 'Previous schedule status',
+ `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
+ `creator` varchar(64) NOT NULL COMMENT 'Creator name',
+ `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
+ `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
+ `modify_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
+ `version` int(11) NOT NULL DEFAULT '1' COMMENT 'Version number, which will be incremented by 1 after modification',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `unique_group_schedule_config` (`inlong_group_id`, `is_deleted`)
+ ) ENGINE = InnoDB
+ DEFAULT CHARSET = utf8mb4 COMMENT = 'schedule_config';
+-- ----------------------------
+
+
SET FOREIGN_KEY_CHECKS = 1;
diff --git a/inlong-manager/manager-web/sql/apache_inlong_manager.sql b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
index cd31161b1c..f7dc1c0a90 100644
--- a/inlong-manager/manager-web/sql/apache_inlong_manager.sql
+++ b/inlong-manager/manager-web/sql/apache_inlong_manager.sql
@@ -1015,5 +1015,33 @@ CREATE TABLE IF NOT EXISTS `tenant_template`
DEFAULT CHARSET = utf8mb4 COMMENT ='Tenant template table';
-- ----------------------------
+-- Table structure for schedule_config
+-- ----------------------------
+CREATE TABLE IF NOT EXISTS `schedule_config`
+(
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
+ `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, undeleted ones cannot be repeated',
+ `schedule_type` int(4) NOT NULL DEFAULT '0' COMMENT 'Schedule type, 0 for normal, 1 for crontab',
+ `schedule_unit` varchar(64) NOT NULL COMMENT 'Schedule unit,M=month, W=week, D=day, H=hour, M=minute, O=oneway',
+ `schedule_interval` int(11) DEFAULT '1' COMMENT 'Schedule interval',
+ `start_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Start time for schedule',
+ `end_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'End time for schedule',
+ `delay_time` int(11) DEFAULT '0' COMMENT 'Delay time in minutes to schedule',
+ `self_depend` int(11) DEFAULT NULL COMMENT 'Self depend info',
+ `task_parallelism` int(11) DEFAULT NULL COMMENT 'Task parallelism',
+ `crontab_expression` varchar(256) DEFAULT NULL COMMENT 'Crontab expression if schedule type is crontab',
+ `status` int(4) DEFAULT '100' COMMENT 'Schedule status',
+ `previous_status` int(4) DEFAULT '100' COMMENT 'Previous schedule status',
+ `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
+ `creator` varchar(64) NOT NULL COMMENT 'Creator name',
+ `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
+ `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
+ `modify_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
+ `version` int(11) NOT NULL DEFAULT '1' COMMENT 'Version number, which will be incremented by 1 after modification',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `unique_group_schedule_config` (`inlong_group_id`, `is_deleted`)
+) ENGINE = InnoDB
+ DEFAULT CHARSET = utf8mb4 COMMENT = 'schedule_config';
+-- ----------------------------
SET FOREIGN_KEY_CHECKS = 1;
diff --git a/inlong-manager/manager-web/sql/changes-1.13.0.sql b/inlong-manager/manager-web/sql/changes-1.13.0.sql
index ecb38b543d..38e7627c5e 100644
--- a/inlong-manager/manager-web/sql/changes-1.13.0.sql
+++ b/inlong-manager/manager-web/sql/changes-1.13.0.sql
@@ -88,3 +88,33 @@ CREATE TABLE IF NOT EXISTS `tenant_template`
UNIQUE KEY `unique_tenant_inlong_template` (`tenant`, `template_name`, `is_deleted`)
) ENGINE = InnoDB
DEFAULT CHARSET = utf8mb4 COMMENT ='Tenant template table';
+
+-- ----------------------------
+-- Table structure for schedule_config
+-- ----------------------------
+CREATE TABLE IF NOT EXISTS `schedule_config`
+(
+ `id` int(11) NOT NULL AUTO_INCREMENT COMMENT 'Incremental primary key',
+ `inlong_group_id` varchar(256) NOT NULL COMMENT 'Inlong group id, undeleted ones cannot be repeated',
+ `schedule_type` int(4) NOT NULL DEFAULT '0' COMMENT 'Schedule type, 0 for normal, 1 for crontab',
+ `schedule_unit` varchar(64) NOT NULL COMMENT 'Schedule unit,M=month, W=week, D=day, H=hour, M=minute, O=oneway',
+ `schedule_interval` int(11) DEFAULT '1' COMMENT 'Schedule interval',
+ `start_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Start time for schedule',
+ `end_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'End time for schedule',
+ `delay_time` int(11) DEFAULT '0' COMMENT 'Delay time in minutes to schedule',
+ `self_depend` int(11) DEFAULT NULL COMMENT 'Self depend info',
+ `task_parallelism` int(11) DEFAULT NULL COMMENT 'Task parallelism',
+ `crontab_expression` varchar(256) DEFAULT NULL COMMENT 'Crontab expression if schedule type is crontab',
+ `status` int(4) DEFAULT '100' COMMENT 'Schedule status',
+ `previous_status` int(4) DEFAULT '100' COMMENT 'Previous schedule status',
+ `is_deleted` int(11) DEFAULT '0' COMMENT 'Whether to delete, 0: not deleted, > 0: deleted',
+ `creator` varchar(64) NOT NULL COMMENT 'Creator name',
+ `modifier` varchar(64) DEFAULT NULL COMMENT 'Modifier name',
+ `create_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT 'Create time',
+ `modify_time` timestamp NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT 'Modify time',
+ `version` int(11) NOT NULL DEFAULT '1' COMMENT 'Version number, which will be incremented by 1 after modification',
+ PRIMARY KEY (`id`),
+ UNIQUE KEY `unique_group_schedule_config` (`inlong_group_id`, `is_deleted`)
+ ) ENGINE = InnoDB
+ DEFAULT CHARSET = utf8mb4 COMMENT = 'schedule_config';
+-- ----------------------------
diff --git a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InLongSchedulerController.java b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InLongSchedulerController.java
new file mode 100644
index 0000000000..801cc09b0e
--- /dev/null
+++ b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InLongSchedulerController.java
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.web.controller;
+
+import org.apache.inlong.manager.common.enums.OperationTarget;
+import org.apache.inlong.manager.common.enums.OperationType;
+import org.apache.inlong.manager.common.validation.UpdateValidation;
+import org.apache.inlong.manager.pojo.common.Response;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
+import org.apache.inlong.manager.pojo.user.LoginUserUtils;
+import org.apache.inlong.manager.service.operationlog.OperationLog;
+import org.apache.inlong.manager.service.schedule.ScheduleService;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiImplicitParam;
+import io.swagger.annotations.ApiOperation;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/api")
+@Api(tags = "Inlong-Schedule-API")
+public class InLongSchedulerController {
+
+ @Autowired
+ private ScheduleService scheduleService;
+
+ @RequestMapping(value = "/schedule/save", method = RequestMethod.POST)
+ @OperationLog(operation = OperationType.CREATE, operationTarget = OperationTarget.SCHEDULE)
+ @ApiOperation(value = "Save schedule info")
+ public Response<Integer> save(@RequestBody ScheduleInfoRequest request) {
+ int result = scheduleService.save(request, LoginUserUtils.getLoginUser().getName());
+ return Response.success(result);
+ }
+
+ @RequestMapping(value = "/schedule/exist/{groupId}", method = RequestMethod.GET)
+ @ApiOperation(value = "Is the schedule info exists for inlong group")
+ @ApiImplicitParam(name = "groupId", dataTypeClass = String.class, required = true)
+ public Response<Boolean> exist(@PathVariable String groupId) {
+ return Response.success(scheduleService.exist(groupId));
+ }
+
+ @RequestMapping(value = "/schedule/get", method = RequestMethod.GET)
+ @ApiOperation(value = "Get schedule info for inlong group")
+ @ApiImplicitParam(name = "groupId", dataTypeClass = String.class, required = true)
+ public Response<ScheduleInfo> get(@RequestParam String groupId) {
+ return Response.success(scheduleService.get(groupId));
+ }
+
+ @RequestMapping(value = "/schedule/update", method = RequestMethod.POST)
+ @OperationLog(operation = OperationType.UPDATE, operationTarget = OperationTarget.SCHEDULE)
+ @ApiOperation(value = "Update schedule info")
+ public Response<Boolean> update(@Validated(UpdateValidation.class) @RequestBody ScheduleInfoRequest request) {
+ return Response.success(scheduleService.update(request, LoginUserUtils.getLoginUser().getName()));
+ }
+
+ @RequestMapping(value = "/schedule/delete/{groupId}", method = RequestMethod.DELETE)
+ @ApiOperation(value = "Delete schedule info")
+ @OperationLog(operation = OperationType.DELETE, operationTarget = OperationTarget.SCHEDULE)
+ @ApiImplicitParam(name = "groupId", value = "Inlong group id", dataTypeClass = String.class, required = true)
+ public Response<Boolean> delete(@PathVariable String groupId) {
+ String operator = LoginUserUtils.getLoginUser().getName();
+ return Response.success(scheduleService.deleteByGroupId(groupId, operator));
+ }
+
+}
(inlong) 08/09: [INLONG-10436][Manager] Move schedule configuration from stream to group (#10437)
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 151c5f4ea2ac4489bcd7f93d4b10f714a8eb99bc
Author: AloysZhang <al...@apache.org>
AuthorDate: Tue Jun 18 12:28:57 2024 +0800
[INLONG-10436][Manager] Move schedule configuration from stream to group (#10437)
---
.../plugin/listener/StartupSortListener.java | 8 ++++
.../manager/pojo/sort/util/StreamParseUtils.java | 22 ----------
.../{ => manager}/schedule/ScheduleEngine.java | 2 +-
.../schedule/ScheduleEngineClient.java | 2 +-
.../{ => manager}/schedule/ScheduleType.java | 2 +-
.../{ => manager}/schedule/ScheduleUnit.java | 2 +-
.../exception/QuartzScheduleException.java | 2 +-
.../schedule/quartz/QuartzOfflineSyncJob.java | 2 +-
.../schedule/quartz/QuartzScheduleClient.java | 4 +-
.../schedule/quartz/QuartzScheduleEngine.java | 10 ++---
.../schedule/quartz/QuartzSchedulerListener.java | 2 +-
.../{ => manager}/schedule/util/ScheduleUtils.java | 10 ++---
.../{ => manager}/schedule/BaseScheduleTest.java | 6 +--
.../{ => manager}/schedule/quartz/MockJob.java | 2 +-
.../schedule/quartz/QuartzScheduleEngineTest.java | 6 +--
.../schedule/util/ScheduleUtilsTest.java | 22 +++++-----
.../service/listener/GroupTaskListenerFactory.java | 20 +++++++++
.../listener/StreamTaskListenerFactory.java | 29 -------------
...ner.java => GroupScheduleResourceListener.java} | 48 ++++++++++------------
.../service/listener/sort/SortConfigListener.java | 10 ++---
.../manager/service/plugin/PluginClassLoader.java | 2 +-
.../group/CreateGroupWorkflowDefinition.java | 11 ++++-
.../stream/CreateStreamWorkflowDefinition.java | 11 +----
.../group/CreateGroupWorkflowDefinitionTest.java | 3 +-
24 files changed, 105 insertions(+), 133 deletions(-)
diff --git a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
index ae2de1c86f..b0624fbe66 100644
--- a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
+++ b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
@@ -21,6 +21,7 @@ import org.apache.inlong.manager.common.consts.InlongConstants;
import org.apache.inlong.manager.common.enums.GroupOperateType;
import org.apache.inlong.manager.common.enums.TaskEvent;
import org.apache.inlong.manager.plugin.util.FlinkUtils;
+import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.pojo.stream.InlongStreamExtInfo;
import org.apache.inlong.manager.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.pojo.workflow.form.process.GroupResourceProcessForm;
@@ -76,6 +77,13 @@ public class StartupSortListener implements SortOperateListener {
}
GroupResourceProcessForm groupResourceForm = (GroupResourceProcessForm) processForm;
+ InlongGroupInfo groupInfo = groupResourceForm.getGroupInfo();
+ // do not build sort config if the group mode is offline
+ if (InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())) {
+ log.info("no need to launching sort job for groupId={} as the mode is offline",
+ groupId);
+ return ListenerResult.success();
+ }
List<InlongStreamInfo> streamInfos = groupResourceForm.getStreamInfos();
int sinkCount = streamInfos.stream()
.map(s -> s.getSinkList() == null ? 0 : s.getSinkList().size())
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/StreamParseUtils.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/StreamParseUtils.java
index 203bcc6778..005529c299 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/StreamParseUtils.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/sort/util/StreamParseUtils.java
@@ -17,14 +17,11 @@
package org.apache.inlong.manager.pojo.sort.util;
-import org.apache.inlong.manager.common.consts.InlongConstants;
import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
import org.apache.inlong.manager.common.enums.TransformType;
import org.apache.inlong.manager.common.util.Preconditions;
import org.apache.inlong.manager.pojo.sink.StreamSink;
import org.apache.inlong.manager.pojo.source.StreamSource;
-import org.apache.inlong.manager.pojo.stream.InlongStreamExtInfo;
-import org.apache.inlong.manager.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.pojo.stream.StreamNode;
import org.apache.inlong.manager.pojo.stream.StreamPipeline;
import org.apache.inlong.manager.pojo.stream.StreamTransform;
@@ -41,8 +38,6 @@ import org.apache.inlong.manager.pojo.transform.splitter.SplitterDefinition;
import com.google.gson.Gson;
import com.google.gson.JsonObject;
-import org.apache.commons.collections4.CollectionUtils;
-import org.apache.commons.lang3.StringUtils;
/**
* Utils of stream parse.
@@ -158,21 +153,4 @@ public class StreamParseUtils {
String.format(" should not be null for streamId=%s", inlongStreamId));
return GSON.fromJson(tempView, StreamPipeline.class);
}
-
- public static String getStreamExtProperty(String key, InlongStreamInfo streamInfo) {
- if (StringUtils.isNotBlank(key) && streamInfo != null && CollectionUtils.isNotEmpty(streamInfo.getExtList())) {
- for (InlongStreamExtInfo ext : streamInfo.getExtList()) {
- if (key.equalsIgnoreCase(ext.getKeyName())) {
- return ext.getKeyValue();
- }
- }
- }
- return null;
- }
-
- public static boolean isRegisterScheduleSuccess(InlongStreamInfo streamInfo) {
- return InlongConstants.REGISTERED
- .equalsIgnoreCase(getStreamExtProperty(InlongConstants.REGISTER_SCHEDULE_STATUS, streamInfo));
- }
-
}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngine.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngine.java
similarity index 97%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngine.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngine.java
index bfdb9a88ad..1f52e280de 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngine.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngine.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule;
+package org.apache.inlong.manager.schedule;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngineClient.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java
similarity index 97%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngineClient.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java
index 1e5ce3460c..dee5cbb2db 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngineClient.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule;
+package org.apache.inlong.manager.schedule;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleType.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleType.java
similarity index 96%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleType.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleType.java
index 0f296fec42..6c64d46e4f 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleType.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleType.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule;
+package org.apache.inlong.manager.schedule;
import lombok.Getter;
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleUnit.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleUnit.java
similarity index 96%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleUnit.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleUnit.java
index 7b4d60779f..f4f0a766d8 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleUnit.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleUnit.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule;
+package org.apache.inlong.manager.schedule;
import lombok.Getter;
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/exception/QuartzScheduleException.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/exception/QuartzScheduleException.java
similarity index 95%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/exception/QuartzScheduleException.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/exception/QuartzScheduleException.java
index 2d0ff005bd..94d466acdf 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/exception/QuartzScheduleException.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/exception/QuartzScheduleException.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule.exception;
+package org.apache.inlong.manager.schedule.exception;
/**
* Exceptions occur in the schedule procedure.
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzOfflineSyncJob.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzOfflineSyncJob.java
similarity index 96%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzOfflineSyncJob.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzOfflineSyncJob.java
index 27628b3320..8ac9d3b6fb 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzOfflineSyncJob.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzOfflineSyncJob.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule.quartz;
+package org.apache.inlong.manager.schedule.quartz;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleClient.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleClient.java
similarity index 93%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleClient.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleClient.java
index 41fc8814a1..05aa6c01ae 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleClient.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleClient.java
@@ -15,10 +15,10 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule.quartz;
+package org.apache.inlong.manager.schedule.quartz;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
-import org.apache.inlong.schedule.ScheduleEngineClient;
+import org.apache.inlong.manager.schedule.ScheduleEngineClient;
/**
* Built-in implementation of schedule engine client corresponding with {@link QuartzScheduleEngine}.
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngine.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngine.java
similarity index 93%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngine.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngine.java
index 31736f1887..d9d2620211 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngine.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngine.java
@@ -15,11 +15,11 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule.quartz;
+package org.apache.inlong.manager.schedule.quartz;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
-import org.apache.inlong.schedule.ScheduleEngine;
-import org.apache.inlong.schedule.exception.QuartzScheduleException;
+import org.apache.inlong.manager.schedule.ScheduleEngine;
+import org.apache.inlong.manager.schedule.exception.QuartzScheduleException;
import com.google.common.annotations.VisibleForTesting;
import lombok.Getter;
@@ -35,8 +35,8 @@ import org.slf4j.LoggerFactory;
import java.util.HashSet;
import java.util.Set;
-import static org.apache.inlong.schedule.util.ScheduleUtils.genQuartzJobDetail;
-import static org.apache.inlong.schedule.util.ScheduleUtils.genQuartzTrigger;
+import static org.apache.inlong.manager.schedule.util.ScheduleUtils.genQuartzJobDetail;
+import static org.apache.inlong.manager.schedule.util.ScheduleUtils.genQuartzTrigger;
/**
* The default implementation of schedule engine based on Quartz scheduler. Response for processing
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzSchedulerListener.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzSchedulerListener.java
similarity index 98%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzSchedulerListener.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzSchedulerListener.java
index ca4f9f1b03..49a9583eaf 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/quartz/QuartzSchedulerListener.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzSchedulerListener.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule.quartz;
+package org.apache.inlong.manager.schedule.quartz;
import org.quartz.JobDetail;
import org.quartz.JobKey;
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/util/ScheduleUtils.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/util/ScheduleUtils.java
similarity index 94%
rename from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/util/ScheduleUtils.java
rename to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/util/ScheduleUtils.java
index 73114fcb6c..65475a0b98 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/util/ScheduleUtils.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/util/ScheduleUtils.java
@@ -15,13 +15,13 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule.util;
+package org.apache.inlong.manager.schedule.util;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
-import org.apache.inlong.schedule.ScheduleType;
-import org.apache.inlong.schedule.ScheduleUnit;
-import org.apache.inlong.schedule.exception.QuartzScheduleException;
-import org.apache.inlong.schedule.quartz.QuartzOfflineSyncJob;
+import org.apache.inlong.manager.schedule.ScheduleType;
+import org.apache.inlong.manager.schedule.ScheduleUnit;
+import org.apache.inlong.manager.schedule.exception.QuartzScheduleException;
+import org.apache.inlong.manager.schedule.quartz.QuartzOfflineSyncJob;
import org.apache.commons.lang3.StringUtils;
import org.quartz.CronScheduleBuilder;
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/BaseScheduleTest.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/BaseScheduleTest.java
similarity index 96%
rename from inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/BaseScheduleTest.java
rename to inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/BaseScheduleTest.java
index c0ccc4b14c..a6eb129d5a 100644
--- a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/BaseScheduleTest.java
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/BaseScheduleTest.java
@@ -15,14 +15,14 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule;
+package org.apache.inlong.manager.schedule;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
-import org.apache.inlong.schedule.exception.QuartzScheduleException;
+import org.apache.inlong.manager.schedule.exception.QuartzScheduleException;
import java.sql.Timestamp;
-import static org.apache.inlong.schedule.ScheduleUnit.SECOND;
+import static org.apache.inlong.manager.schedule.ScheduleUnit.SECOND;
public class BaseScheduleTest {
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/MockJob.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/MockJob.java
similarity index 97%
rename from inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/MockJob.java
rename to inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/MockJob.java
index 47ce19ad09..9202ea5b40 100644
--- a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/MockJob.java
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/MockJob.java
@@ -15,7 +15,7 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule.quartz;
+package org.apache.inlong.manager.schedule.quartz;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngineTest.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngineTest.java
similarity index 97%
rename from inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngineTest.java
rename to inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngineTest.java
index fc5ca9d9fd..008a7e42f8 100644
--- a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/quartz/QuartzScheduleEngineTest.java
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngineTest.java
@@ -15,10 +15,10 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule.quartz;
+package org.apache.inlong.manager.schedule.quartz;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
-import org.apache.inlong.schedule.BaseScheduleTest;
+import org.apache.inlong.manager.schedule.BaseScheduleTest;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -27,7 +27,7 @@ import org.quartz.JobKey;
import java.util.concurrent.TimeUnit;
-import static org.apache.inlong.schedule.ScheduleUnit.SECOND;
+import static org.apache.inlong.manager.schedule.ScheduleUnit.SECOND;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/util/ScheduleUtilsTest.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/util/ScheduleUtilsTest.java
similarity index 89%
rename from inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/util/ScheduleUtilsTest.java
rename to inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/util/ScheduleUtilsTest.java
index 77b84f3064..415331be3d 100644
--- a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/schedule/util/ScheduleUtilsTest.java
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/util/ScheduleUtilsTest.java
@@ -15,12 +15,12 @@
* limitations under the License.
*/
-package org.apache.inlong.schedule.util;
+package org.apache.inlong.manager.schedule.util;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
-import org.apache.inlong.schedule.BaseScheduleTest;
-import org.apache.inlong.schedule.exception.QuartzScheduleException;
-import org.apache.inlong.schedule.quartz.QuartzOfflineSyncJob;
+import org.apache.inlong.manager.schedule.BaseScheduleTest;
+import org.apache.inlong.manager.schedule.exception.QuartzScheduleException;
+import org.apache.inlong.manager.schedule.quartz.QuartzOfflineSyncJob;
import org.junit.jupiter.api.Test;
import org.quartz.CronScheduleBuilder;
@@ -35,13 +35,13 @@ import org.quartz.TriggerKey;
import java.util.Date;
-import static org.apache.inlong.schedule.ScheduleUnit.DAY;
-import static org.apache.inlong.schedule.ScheduleUnit.HOUR;
-import static org.apache.inlong.schedule.ScheduleUnit.MINUTE;
-import static org.apache.inlong.schedule.ScheduleUnit.MONTH;
-import static org.apache.inlong.schedule.ScheduleUnit.ONE_WAY;
-import static org.apache.inlong.schedule.ScheduleUnit.WEEK;
-import static org.apache.inlong.schedule.ScheduleUnit.YEAR;
+import static org.apache.inlong.manager.schedule.ScheduleUnit.DAY;
+import static org.apache.inlong.manager.schedule.ScheduleUnit.HOUR;
+import static org.apache.inlong.manager.schedule.ScheduleUnit.MINUTE;
+import static org.apache.inlong.manager.schedule.ScheduleUnit.MONTH;
+import static org.apache.inlong.manager.schedule.ScheduleUnit.ONE_WAY;
+import static org.apache.inlong.manager.schedule.ScheduleUnit.WEEK;
+import static org.apache.inlong.manager.schedule.ScheduleUnit.YEAR;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/GroupTaskListenerFactory.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/GroupTaskListenerFactory.java
index a690472304..3ff9942c5d 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/GroupTaskListenerFactory.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/GroupTaskListenerFactory.java
@@ -20,6 +20,7 @@ package org.apache.inlong.manager.service.listener;
import org.apache.inlong.manager.common.plugin.Plugin;
import org.apache.inlong.manager.common.plugin.PluginBinder;
import org.apache.inlong.manager.service.listener.queue.QueueResourceListener;
+import org.apache.inlong.manager.service.listener.schedule.GroupScheduleResourceListener;
import org.apache.inlong.manager.service.listener.sink.SinkResourceListener;
import org.apache.inlong.manager.service.listener.sort.SortConfigListener;
import org.apache.inlong.manager.service.listener.source.SourceDeleteListener;
@@ -29,6 +30,7 @@ import org.apache.inlong.manager.workflow.WorkflowContext;
import org.apache.inlong.manager.workflow.definition.ServiceTaskType;
import org.apache.inlong.manager.workflow.definition.TaskListenerFactory;
import org.apache.inlong.manager.workflow.event.task.QueueOperateListener;
+import org.apache.inlong.manager.workflow.event.task.ScheduleOperateListener;
import org.apache.inlong.manager.workflow.event.task.SortOperateListener;
import org.apache.inlong.manager.workflow.event.task.SourceOperateListener;
import org.apache.inlong.manager.workflow.event.task.TaskEventListener;
@@ -57,6 +59,7 @@ public class GroupTaskListenerFactory implements PluginBinder, TaskListenerFacto
private List<SourceOperateListener> sourceOperateListeners;
private List<QueueOperateListener> queueOperateListeners;
private List<SortOperateListener> sortOperateListeners;
+ private List<ScheduleOperateListener> scheduleOperateListeners;
@Autowired
private SourceStopListener sourceStopListener;
@@ -70,6 +73,8 @@ public class GroupTaskListenerFactory implements PluginBinder, TaskListenerFacto
private SinkResourceListener sinkResourceListener;
@Autowired
private SortConfigListener sortConfigListener;
+ @Autowired
+ private GroupScheduleResourceListener groupScheduleResourceListener;
@PostConstruct
public void init() {
@@ -81,6 +86,8 @@ public class GroupTaskListenerFactory implements PluginBinder, TaskListenerFacto
queueOperateListeners.add(queueResourceListener);
sortOperateListeners = new LinkedList<>();
sortOperateListeners.add(sortConfigListener);
+ scheduleOperateListeners = new LinkedList<>();
+ scheduleOperateListeners.add(groupScheduleResourceListener);
}
@Override
@@ -124,6 +131,9 @@ public class GroupTaskListenerFactory implements PluginBinder, TaskListenerFacto
return Lists.newArrayList(sourceOperateListeners);
case INIT_SINK:
return Collections.singletonList(sinkResourceListener);
+ case INIT_SCHEDULE:
+ List<ScheduleOperateListener> scheduleOperateListeners = getScheduleOperateListener(workflowContext);
+ return Lists.newArrayList(scheduleOperateListeners);
default:
throw new IllegalArgumentException(String.format("Unsupported ServiceTaskType %s", taskType));
}
@@ -177,4 +187,14 @@ public class GroupTaskListenerFactory implements PluginBinder, TaskListenerFacto
return listeners;
}
+ public List<ScheduleOperateListener> getScheduleOperateListener(WorkflowContext context) {
+ List<ScheduleOperateListener> listeners = new ArrayList<>();
+ for (ScheduleOperateListener listener : scheduleOperateListeners) {
+ if (listener != null && listener.accept(context)) {
+ listeners.add(listener);
+ }
+ }
+ return listeners;
+ }
+
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/StreamTaskListenerFactory.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/StreamTaskListenerFactory.java
index 87c3a867c2..5069423c13 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/StreamTaskListenerFactory.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/StreamTaskListenerFactory.java
@@ -20,14 +20,12 @@ package org.apache.inlong.manager.service.listener;
import org.apache.inlong.manager.common.plugin.Plugin;
import org.apache.inlong.manager.common.plugin.PluginBinder;
import org.apache.inlong.manager.service.listener.queue.StreamQueueResourceListener;
-import org.apache.inlong.manager.service.listener.schedule.StreamScheduleResourceListener;
import org.apache.inlong.manager.service.listener.sink.StreamSinkResourceListener;
import org.apache.inlong.manager.service.listener.sort.StreamSortConfigListener;
import org.apache.inlong.manager.workflow.WorkflowContext;
import org.apache.inlong.manager.workflow.definition.ServiceTaskType;
import org.apache.inlong.manager.workflow.definition.TaskListenerFactory;
import org.apache.inlong.manager.workflow.event.task.QueueOperateListener;
-import org.apache.inlong.manager.workflow.event.task.ScheduleOperateListener;
import org.apache.inlong.manager.workflow.event.task.SinkOperateListener;
import org.apache.inlong.manager.workflow.event.task.SortOperateListener;
import org.apache.inlong.manager.workflow.event.task.SourceOperateListener;
@@ -55,7 +53,6 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
private List<QueueOperateListener> queueOperateListeners;
private List<SortOperateListener> sortOperateListeners;
private List<SinkOperateListener> sinkOperateListeners;
- private List<ScheduleOperateListener> scheduleOperateListeners;
@Autowired
private StreamQueueResourceListener queueResourceListener;
@@ -63,8 +60,6 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
private StreamSortConfigListener streamSortConfigListener;
@Autowired
private StreamSinkResourceListener sinkResourceListener;
- @Autowired
- private StreamScheduleResourceListener scheduleResourceListener;
@PostConstruct
public void init() {
@@ -75,8 +70,6 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
sortOperateListeners.add(streamSortConfigListener);
sinkOperateListeners = new LinkedList<>();
sinkOperateListeners.add(sinkResourceListener);
- scheduleOperateListeners = new LinkedList<>();
- scheduleOperateListeners.add(scheduleResourceListener);
}
@Override
@@ -101,10 +94,6 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
if (CollectionUtils.isNotEmpty(pluginSinkOperateListeners)) {
sinkOperateListeners.addAll(pluginSinkOperateListeners);
}
- List<ScheduleOperateListener> pluginScheduleOperateListeners = processPlugin.createScheduleOperateListeners();
- if (CollectionUtils.isNotEmpty(pluginScheduleOperateListeners)) {
- scheduleOperateListeners.addAll(pluginScheduleOperateListeners);
- }
}
@Override
@@ -129,9 +118,6 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
case INIT_SINK:
List<SinkOperateListener> sinkOperateListeners = getSinkOperateListener(workflowContext);
return Lists.newArrayList(sinkOperateListeners);
- case INIT_SCHEDULE:
- List<ScheduleOperateListener> scheduleOperateListeners = getScheduleOperateListener(workflowContext);
- return Lists.newArrayList(scheduleOperateListeners);
default:
throw new IllegalArgumentException(String.format("Unsupported ServiceTaskType %s", taskType));
}
@@ -145,7 +131,6 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
queueOperateListeners = new LinkedList<>();
sortOperateListeners = new LinkedList<>();
sinkOperateListeners = new LinkedList<>();
- scheduleOperateListeners = new LinkedList<>();
}
/**
@@ -199,18 +184,4 @@ public class StreamTaskListenerFactory implements PluginBinder, TaskListenerFact
}
return listeners;
}
-
- /**
- * Get schedule operate listener list.
- */
- private List<ScheduleOperateListener> getScheduleOperateListener(WorkflowContext context) {
- List<ScheduleOperateListener> listeners = new ArrayList<>();
- for (ScheduleOperateListener listener : scheduleOperateListeners) {
- if (listener != null && listener.accept(context)) {
- listeners.add(listener);
- }
- }
- return listeners;
- }
-
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/StreamScheduleResourceListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/GroupScheduleResourceListener.java
similarity index 54%
rename from inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/StreamScheduleResourceListener.java
rename to inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/GroupScheduleResourceListener.java
index 61179dc343..e27dab9e89 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/StreamScheduleResourceListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/GroupScheduleResourceListener.java
@@ -20,10 +20,10 @@ package org.apache.inlong.manager.service.listener.schedule;
import org.apache.inlong.manager.common.consts.InlongConstants;
import org.apache.inlong.manager.common.enums.GroupOperateType;
import org.apache.inlong.manager.common.enums.TaskEvent;
-import org.apache.inlong.manager.pojo.stream.InlongStreamExtInfo;
-import org.apache.inlong.manager.pojo.stream.InlongStreamInfo;
+import org.apache.inlong.manager.pojo.group.InlongGroupExtInfo;
+import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
+import org.apache.inlong.manager.pojo.workflow.form.process.GroupResourceProcessForm;
import org.apache.inlong.manager.pojo.workflow.form.process.ProcessForm;
-import org.apache.inlong.manager.pojo.workflow.form.process.StreamResourceProcessForm;
import org.apache.inlong.manager.workflow.WorkflowContext;
import org.apache.inlong.manager.workflow.event.ListenerResult;
import org.apache.inlong.manager.workflow.event.task.ScheduleOperateListener;
@@ -35,7 +35,7 @@ import java.util.List;
@Service
@Slf4j
-public class StreamScheduleResourceListener implements ScheduleOperateListener {
+public class GroupScheduleResourceListener implements ScheduleOperateListener {
@Override
public TaskEvent event() {
@@ -46,48 +46,44 @@ public class StreamScheduleResourceListener implements ScheduleOperateListener {
public boolean accept(WorkflowContext context) {
ProcessForm processForm = context.getProcessForm();
String groupId = processForm.getInlongGroupId();
- if (!(processForm instanceof StreamResourceProcessForm)) {
- log.info("not add schedule stream listener, not StreamResourceProcessForm for groupId [{}]", groupId);
+ if (!(processForm instanceof GroupResourceProcessForm)) {
+ log.info("not add schedule group listener, not groupResourceProcessForm for groupId [{}]", groupId);
return false;
}
- StreamResourceProcessForm streamProcessForm = (StreamResourceProcessForm) processForm;
- String streamId = streamProcessForm.getStreamInfo().getInlongStreamId();
- if (streamProcessForm.getGroupOperateType() != GroupOperateType.INIT) {
- log.info("not add schedule stream listener, as the operate was not INIT for groupId [{}] streamId [{}]",
- groupId, streamId);
+ GroupResourceProcessForm groupProcessForm = (GroupResourceProcessForm) processForm;
+ if (groupProcessForm.getGroupOperateType() != GroupOperateType.INIT) {
+ log.info("not add schedule group listener, as the operate was not INIT for groupId [{}]", groupId);
return false;
}
- log.info("add startup stream listener for groupId [{}] streamId [{}]", groupId, streamId);
- return InlongConstants.DATASYNC_OFFLINE_MODE.equals(streamProcessForm.getGroupInfo().getInlongGroupMode());
+ log.info("add startup group listener for groupId [{}]", groupId);
+ return InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupProcessForm.getGroupInfo().getInlongGroupMode());
}
@Override
public ListenerResult listen(WorkflowContext context) throws Exception {
- StreamResourceProcessForm form = (StreamResourceProcessForm) context.getProcessForm();
- InlongStreamInfo streamInfo = form.getStreamInfo();
- final String groupId = streamInfo.getInlongGroupId();
- final String streamId = streamInfo.getInlongStreamId();
- log.info("begin to register schedule info for groupId={}, streamId={}", groupId, streamId);
+ GroupResourceProcessForm form = (GroupResourceProcessForm) context.getProcessForm();
+ InlongGroupInfo groupInfo = form.getGroupInfo();
+ final String groupId = groupInfo.getInlongGroupId();
+ log.info("begin to register schedule info for groupId={}", groupId);
// todo: register schedule info to schedule service
- // after register schedule info successfully, add ext property to stream info
- saveInfo(streamInfo, InlongConstants.REGISTER_SCHEDULE_STATUS,
- InlongConstants.REGISTERED, streamInfo.getExtList());
- log.info("success to register schedule info for group [" + groupId + "] and stream [" + streamId + "]");
+ // after register schedule info successfully, add ext property to group ext info
+ saveInfo(groupInfo, InlongConstants.REGISTER_SCHEDULE_STATUS,
+ InlongConstants.REGISTERED, groupInfo.getExtList());
+ log.info("success to register schedule info for group={}", groupId);
return ListenerResult.success();
}
/**
* Save stream ext info into list.
*/
- private void saveInfo(InlongStreamInfo streamInfo, String keyName, String keyValue,
- List<InlongStreamExtInfo> extInfoList) {
- InlongStreamExtInfo extInfo = new InlongStreamExtInfo();
+ private void saveInfo(InlongGroupInfo streamInfo, String keyName, String keyValue,
+ List<InlongGroupExtInfo> extInfoList) {
+ InlongGroupExtInfo extInfo = new InlongGroupExtInfo();
extInfo.setInlongGroupId(streamInfo.getInlongGroupId());
- extInfo.setInlongStreamId(streamInfo.getInlongStreamId());
extInfo.setKeyName(keyName);
extInfo.setKeyValue(keyValue);
extInfoList.add(extInfo);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java
index f8981bf09a..067a320a49 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/sort/SortConfigListener.java
@@ -24,7 +24,6 @@ import org.apache.inlong.manager.common.enums.TaskEvent;
import org.apache.inlong.manager.common.exceptions.WorkflowListenerException;
import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.pojo.sink.StreamSink;
-import org.apache.inlong.manager.pojo.sort.util.StreamParseUtils;
import org.apache.inlong.manager.pojo.stream.InlongStreamInfo;
import org.apache.inlong.manager.pojo.workflow.form.process.GroupResourceProcessForm;
import org.apache.inlong.manager.pojo.workflow.form.process.ProcessForm;
@@ -126,11 +125,10 @@ public class SortConfigListener implements SortOperateListener {
try {
for (InlongStreamInfo streamInfo : streamInfos) {
- // do not build sort config if the group mode is offline and the stream is not config successfully
- if (InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())
- && !StreamParseUtils.isRegisterScheduleSuccess(streamInfo)) {
- LOGGER.info("no need to build sort config for groupId={} streamId={} as the mode is offline "
- + "and the stream is not config successfully yet", groupId, streamInfo.getInlongStreamId());
+ // do not build sort config if the group mode is offline
+ if (InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())) {
+ LOGGER.info("no need to build sort config for groupId={} streamId={} as the mode is offline",
+ groupId, streamInfo.getInlongStreamId());
continue;
}
List<StreamSink> sinkList = streamInfo.getSinkList();
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/plugin/PluginClassLoader.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/plugin/PluginClassLoader.java
index 08d04b58ee..1b30839308 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/plugin/PluginClassLoader.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/plugin/PluginClassLoader.java
@@ -88,7 +88,7 @@ public class PluginClassLoader extends URLClassLoader {
* Get pluginClassLoader by plugin url.
*/
public static PluginClassLoader getFromPluginUrl(String url, ClassLoader parent) {
- log.info("ClassLoaderPath:{}", url);
+ log.info("ClassLoaderPath : {}", url);
checkClassLoader(parent);
checkUrl(url);
return AccessController.doPrivileged(new PrivilegedAction<PluginClassLoader>() {
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/group/CreateGroupWorkflowDefinition.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/group/CreateGroupWorkflowDefinition.java
index bc8f0c2546..b9cacdffc4 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/group/CreateGroupWorkflowDefinition.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/group/CreateGroupWorkflowDefinition.java
@@ -86,6 +86,14 @@ public class CreateGroupWorkflowDefinition implements WorkflowDefinition {
initSortTask.setListenerFactory(groupTaskListenerFactory);
process.addTask(initSortTask);
+ // Init Schedule
+ ServiceTask initScheduleTask = new ServiceTask();
+ initScheduleTask.setName("InitSchedule");
+ initScheduleTask.setDisplayName("Group-InitSchedule");
+ initScheduleTask.setServiceTaskType(ServiceTaskType.INIT_SCHEDULE);
+ initScheduleTask.setListenerFactory(groupTaskListenerFactory);
+ process.addTask(initScheduleTask);
+
// End node
EndEvent endEvent = new EndEvent();
process.setEndEvent(endEvent);
@@ -94,7 +102,8 @@ public class CreateGroupWorkflowDefinition implements WorkflowDefinition {
// To ensure that after some tasks fail, data will not start to be collected by source or consumed by sort
startEvent.addNext(initMQTask);
initMQTask.addNext(initSortTask);
- initSortTask.addNext(endEvent);
+ initSortTask.addNext(initScheduleTask);
+ initScheduleTask.addNext(endEvent);
return process;
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java
index 5505ed0694..2ec6e9052e 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/workflow/stream/CreateStreamWorkflowDefinition.java
@@ -102,14 +102,6 @@ public class CreateStreamWorkflowDefinition implements WorkflowDefinition {
initSourceTask.setListenerFactory(streamTaskListenerFactory);
process.addTask(initSourceTask);
- // Init Schedule info
- ServiceTask initScheduleTask = new ServiceTask();
- initScheduleTask.setName("InitSchedule");
- initScheduleTask.setDisplayName("Stream-InitSchedule");
- initScheduleTask.setServiceTaskType(ServiceTaskType.INIT_SCHEDULE);
- initScheduleTask.setListenerFactory(streamTaskListenerFactory);
- process.addTask(initScheduleTask);
-
// End node
EndEvent endEvent = new EndEvent();
process.setEndEvent(endEvent);
@@ -120,8 +112,7 @@ public class CreateStreamWorkflowDefinition implements WorkflowDefinition {
initMQTask.addNext(initSinkTask);
initSinkTask.addNext(initSortTask);
initSortTask.addNext(initSourceTask);
- initSourceTask.addNext(initScheduleTask);
- initScheduleTask.addNext(endEvent);
+ initSourceTask.addNext(endEvent);
return process;
}
diff --git a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/group/CreateGroupWorkflowDefinitionTest.java b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/group/CreateGroupWorkflowDefinitionTest.java
index abae87fef4..e56b013d57 100644
--- a/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/group/CreateGroupWorkflowDefinitionTest.java
+++ b/inlong-manager/manager-service/src/test/java/org/apache/inlong/manager/service/workflow/group/CreateGroupWorkflowDefinitionTest.java
@@ -40,7 +40,8 @@ public class CreateGroupWorkflowDefinitionTest extends ServiceBaseTest {
Assertions.assertNotSame(cloneProcess2, cloneProcess1);
Assertions.assertNotNull(process.getTaskByName("InitMQ"));
Assertions.assertNotNull(process.getTaskByName("InitSort"));
- Assertions.assertEquals(2, process.getNameToTaskMap().size());
+ Assertions.assertNotNull(process.getTaskByName("InitSchedule"));
+ Assertions.assertEquals(3, process.getNameToTaskMap().size());
}
}
(inlong) 09/09: [INLONG-10360][Manager] Combine schedule state transition with group operations (#10445)
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit c1bda980b61181fc8c7dce1880fcc60f5bbbfad3
Author: AloysZhang <al...@apache.org>
AuthorDate: Wed Jun 19 15:41:34 2024 +0800
[INLONG-10360][Manager] Combine schedule state transition with group operations (#10445)
Co-authored-by: fuweng11 <76...@users.noreply.github.com>
---
.../inlong/manager/common/enums/ErrorCodeEnum.java | 2 +
.../manager/common/enums/ScheduleStatus.java | 16 +-
.../inlong/manager/pojo/group/InlongGroupInfo.java | 31 ++++
.../manager/pojo/group/InlongGroupRequest.java | 31 ++++
.../inlong/manager/pojo/schedule/ScheduleInfo.java | 29 ++++
.../manager/pojo/schedule/ScheduleInfoRequest.java | 28 ++++
...leEngineClient.java => NoopScheduleClient.java} | 45 +++---
.../manager/schedule/ScheduleClientFactory.java | 55 +++++++
.../inlong/manager/schedule/ScheduleEngine.java | 4 +-
.../manager/schedule/ScheduleEngineClient.java | 12 +-
.../manager/schedule/ScheduleEngineType.java} | 28 +---
.../schedule/quartz/QuartzScheduleClient.java | 17 +-
.../schedule/quartz/QuartzScheduleEngine.java | 21 +--
.../manager/schedule/util/ScheduleUtils.java | 4 +-
.../inlong/manager/schedule/quartz/MockJob.java | 3 +-
.../schedule/quartz/QuartzScheduleEngineTest.java | 2 +-
.../manager/schedule/util/ScheduleUtilsTest.java | 8 +-
inlong-manager/manager-service/pom.xml | 5 +
.../service/group/InlongGroupServiceImpl.java | 57 ++++++-
.../schedule/GroupScheduleResourceListener.java | 8 +-
.../manager/service/schedule/ScheduleOperator.java | 95 +++++++++++
.../service/schedule/ScheduleOperatorImpl.java | 175 +++++++++++++++++++++
.../manager/service/schedule/ScheduleService.java | 11 ++
.../service/schedule/ScheduleServiceImpl.java | 54 ++++++-
.../web/controller/InLongSchedulerController.java | 24 ++-
.../src/main/resources/application-dev.properties | 6 +-
.../src/main/resources/application-prod.properties | 3 +
.../src/main/resources/application-test.properties | 3 +
.../src/main/resources/application.properties | 4 +
29 files changed, 698 insertions(+), 83 deletions(-)
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java
index 6a8f9b4699..2a82325ad9 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ErrorCodeEnum.java
@@ -127,6 +127,8 @@ public enum ErrorCodeEnum {
SCHEDULE_NOT_FOUND(1700, "Schedule info not found"),
SCHEDULE_DUPLICATE(1701, "Schedule info already exist"),
+ SCHEDULE_ENGINE_NOT_SUPPORTED(1702, "Schedule engine type not supported"),
+ SCHEDULE_STATUS_TRANSITION_NOT_ALLOWED(1703, "Schedule status transition is not allowed"),
WORKFLOW_EXE_FAILED(4000, "Workflow execution exception"),
WORKFLOW_APPROVER_NOT_FOUND(4001, "Workflow approver does not exist/no operation authority"),
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java
index cb256a491c..2d936b5532 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java
@@ -19,11 +19,25 @@ package org.apache.inlong.manager.common.enums;
import lombok.Getter;
+/**
+ * Status for schedule info.
+ * This is the transient status of the schedule info.
+ * With specified operations, the status will change to corresponding value.
+ * Status Operations
+ * NEW inlong group created with schedule info
+ * APPROVED the new inlong group approved by admin
+ * REGISTERED schedule info registered to schedule engine
+ * UPDATED update schedule info for a group
+ * DELETED delete a group
+ * */
@Getter
public enum ScheduleStatus {
NEW(100, "new"),
- DELETED(40, "deleted");
+ APPROVED(101, "approved"),
+ REGISTERED(102, "registered"),
+ UPDATED(103, "updated"),
+ DELETED(99, "deleted");
private final Integer code;
private final String description;
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupInfo.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupInfo.java
index 17b66497a9..0fb0479196 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupInfo.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupInfo.java
@@ -30,6 +30,7 @@ import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.NoArgsConstructor;
+import java.sql.Timestamp;
import java.util.Date;
import java.util.List;
@@ -137,6 +138,36 @@ public abstract class InlongGroupInfo extends BaseInlongGroup {
@ApiModelProperty(value = "Inlong tenant")
private String tenant;
+ // schedule type, support [normal, crontab], 0 for normal and 1 for crontab
+ @ApiModelProperty("Schedule type")
+ private Integer scheduleType;
+
+ // time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
+ // Y=year, M=month, W=week, D=day, H=hour, I=minute, O=oneway
+ @ApiModelProperty("TimeUnit for schedule interval")
+ private String scheduleUnit;
+
+ @ApiModelProperty("Schedule interval")
+ private Integer scheduleInterval;
+
+ @ApiModelProperty("Start time")
+ private Timestamp startTime;
+
+ @ApiModelProperty("End time")
+ private Timestamp endTime;
+
+ @ApiModelProperty("Delay time")
+ private Integer delayTime;
+
+ @ApiModelProperty("Self depend")
+ private Integer selfDepend;
+
+ @ApiModelProperty("Schedule task parallelism")
+ private Integer taskParallelism;
+
+ @ApiModelProperty("Schedule task parallelism")
+ private Integer crontabExpression;
+
public abstract InlongGroupRequest genRequest();
}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupRequest.java
index 6140bddad5..1adc210a0a 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupRequest.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupRequest.java
@@ -34,6 +34,7 @@ import javax.validation.constraints.NotBlank;
import javax.validation.constraints.NotNull;
import javax.validation.constraints.Pattern;
+import java.sql.Timestamp;
import java.util.List;
/**
@@ -130,4 +131,34 @@ public abstract class InlongGroupRequest extends BaseInlongGroup {
@NotNull(groups = UpdateValidation.class, message = "version cannot be null")
private Integer version;
+ // schedule type, support [normal, crontab], 0 for normal and 1 for crontab
+ @ApiModelProperty("Schedule type")
+ private Integer scheduleType;
+
+ // time unit for offline task schedule interval, support [month, week, day, hour, minute, oneway]
+ // Y=year, M=month, W=week, D=day, H=hour, I=minute, O=oneway
+ @ApiModelProperty("TimeUnit for schedule interval")
+ private String scheduleUnit;
+
+ @ApiModelProperty("Schedule interval")
+ private Integer scheduleInterval;
+
+ @ApiModelProperty("Start time")
+ private Timestamp startTime;
+
+ @ApiModelProperty("End time")
+ private Timestamp endTime;
+
+ @ApiModelProperty("Delay time")
+ private Integer delayTime;
+
+ @ApiModelProperty("Self depend")
+ private Integer selfDepend;
+
+ @ApiModelProperty("Schedule task parallelism")
+ private Integer taskParallelism;
+
+ @ApiModelProperty("Schedule task parallelism")
+ private Integer crontabExpression;
+
}
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java
index 2386d817bb..24f2e6196e 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfo.java
@@ -29,6 +29,7 @@ import lombok.NoArgsConstructor;
import javax.validation.constraints.NotNull;
import java.sql.Timestamp;
+import java.util.Objects;
@Data
@Builder
@@ -79,4 +80,32 @@ public class ScheduleInfo {
@NotNull(groups = UpdateValidation.class, message = "version cannot be null")
private Integer version;
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ScheduleInfo that = (ScheduleInfo) o;
+ return Objects.equals(id, that.id) && Objects.equals(inlongGroupId, that.inlongGroupId)
+ && Objects.equals(scheduleType, that.scheduleType) && Objects.equals(scheduleUnit,
+ that.scheduleUnit)
+ && Objects.equals(scheduleInterval, that.scheduleInterval)
+ && Objects.equals(startTime, that.startTime) && Objects.equals(endTime, that.endTime)
+ && Objects.equals(delayTime, that.delayTime) && Objects.equals(selfDepend,
+ that.selfDepend)
+ && Objects.equals(taskParallelism, that.taskParallelism)
+ && Objects.equals(crontabExpression, that.crontabExpression) && Objects.equals(version,
+ that.version);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(id, inlongGroupId, scheduleType, scheduleUnit, scheduleInterval, startTime, endTime,
+ delayTime,
+ selfDepend, taskParallelism, crontabExpression, version);
+ }
+
}
\ No newline at end of file
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java
index b3c117da9a..a324cee7f5 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/schedule/ScheduleInfoRequest.java
@@ -26,6 +26,7 @@ import lombok.Data;
import javax.validation.constraints.NotNull;
import java.sql.Timestamp;
+import java.util.Objects;
@Data
@ApiModel("Schedule request")
@@ -73,4 +74,31 @@ public class ScheduleInfoRequest {
@NotNull(groups = UpdateValidation.class, message = "version cannot be null")
private Integer version;
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ScheduleInfoRequest that = (ScheduleInfoRequest) o;
+ return Objects.equals(id, that.id) && Objects.equals(inlongGroupId, that.inlongGroupId)
+ && Objects.equals(scheduleType, that.scheduleType) && Objects.equals(scheduleUnit,
+ that.scheduleUnit)
+ && Objects.equals(scheduleInterval, that.scheduleInterval)
+ && Objects.equals(startTime, that.startTime) && Objects.equals(endTime, that.endTime)
+ && Objects.equals(delayTime, that.delayTime) && Objects.equals(selfDepend,
+ that.selfDepend)
+ && Objects.equals(taskParallelism, that.taskParallelism)
+ && Objects.equals(crontabExpression, that.crontabExpression) && Objects.equals(version,
+ that.version);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(id, inlongGroupId, scheduleType, scheduleUnit, scheduleInterval, startTime, endTime,
+ delayTime,
+ selfDepend, taskParallelism, crontabExpression, version);
+ }
}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/NoopScheduleClient.java
similarity index 58%
copy from inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java
copy to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/NoopScheduleClient.java
index dee5cbb2db..a122235de3 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/NoopScheduleClient.java
@@ -19,27 +19,28 @@ package org.apache.inlong.manager.schedule;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
-/**
- * Interface for schedule engine client which responses for communicating with schedule engine.
- * */
-public interface ScheduleEngineClient {
-
- /**
- * Register schedule to schedule engine.
- * @param scheduleInfo schedule info to register
- * */
- boolean register(ScheduleInfo scheduleInfo);
-
- /**
- * Un-register schedule from schedule engine.
- * @param scheduleInfo schedule info to unregister
- * */
- boolean unregister(ScheduleInfo scheduleInfo);
-
- /**
- * Update schedule from schedule engine.
- * @param scheduleInfo schedule info to update
- * */
- boolean update(ScheduleInfo scheduleInfo);
+import org.springframework.stereotype.Service;
+@Service
+public class NoopScheduleClient implements ScheduleEngineClient {
+
+ @Override
+ public boolean accept(String engineType) {
+ return ScheduleEngineType.NONE.getType().equals(engineType);
+ }
+
+ @Override
+ public boolean register(ScheduleInfo scheduleInfo) {
+ return true;
+ }
+
+ @Override
+ public boolean unregister(String groupId) {
+ return true;
+ }
+
+ @Override
+ public boolean update(ScheduleInfo scheduleInfo) {
+ return true;
+ }
}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleClientFactory.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleClientFactory.java
new file mode 100644
index 0000000000..13f87b3c45
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleClientFactory.java
@@ -0,0 +1,55 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.schedule;
+
+import org.apache.inlong.manager.common.enums.ErrorCodeEnum;
+import org.apache.inlong.manager.common.exceptions.BusinessException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.Optional;
+
+@Service
+public class ScheduleClientFactory {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ScheduleClientFactory.class);
+
+ @Value("${inlong.schedule.engine:none}")
+ private String scheduleEngineName;
+
+ @Autowired
+ List<ScheduleEngineClient> scheduleEngineClients;
+
+ public ScheduleEngineClient getInstance() {
+ Optional<ScheduleEngineClient> optScheduleClient =
+ scheduleEngineClients.stream().filter(t -> t.accept(scheduleEngineName)).findFirst();
+ if (!optScheduleClient.isPresent()) {
+ LOGGER.warn("Schedule engine client not found for {} ", scheduleEngineName);
+ throw new BusinessException(ErrorCodeEnum.SCHEDULE_ENGINE_NOT_SUPPORTED,
+ String.format(ErrorCodeEnum.SCHEDULE_ENGINE_NOT_SUPPORTED.getMessage(), scheduleEngineName));
+ }
+ LOGGER.info("Get schedule engine client success for {}", scheduleEngineName);
+ return optScheduleClient.get();
+ }
+
+}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngine.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngine.java
index 1f52e280de..bc0c963b38 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngine.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngine.java
@@ -37,9 +37,9 @@ public interface ScheduleEngine {
/**
* Handle schedule unregister.
- * @param scheduleInfo schedule info to unregister
+ * @param groupId group to un-register schedule info
* */
- boolean handleUnregister(ScheduleInfo scheduleInfo);
+ boolean handleUnregister(String groupId);
/**
* Handle schedule update.
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java
index dee5cbb2db..9c6cf081d4 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineClient.java
@@ -24,6 +24,11 @@ import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
* */
public interface ScheduleEngineClient {
+ /**
+ * Check whether scheduleEngine type is matched.
+ * */
+ boolean accept(String engineType);
+
/**
* Register schedule to schedule engine.
* @param scheduleInfo schedule info to register
@@ -32,9 +37,10 @@ public interface ScheduleEngineClient {
/**
* Un-register schedule from schedule engine.
- * @param scheduleInfo schedule info to unregister
- * */
- boolean unregister(ScheduleInfo scheduleInfo);
+ *
+ * @param groupId schedule info to unregister
+ */
+ boolean unregister(String groupId);
/**
* Update schedule from schedule engine.
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineType.java
similarity index 57%
copy from inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java
copy to inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineType.java
index cb256a491c..71949ef744 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/ScheduleStatus.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/ScheduleEngineType.java
@@ -15,31 +15,19 @@
* limitations under the License.
*/
-package org.apache.inlong.manager.common.enums;
+package org.apache.inlong.manager.schedule;
import lombok.Getter;
@Getter
-public enum ScheduleStatus {
+public enum ScheduleEngineType {
- NEW(100, "new"),
- DELETED(40, "deleted");
+ NONE("None"),
+ QUARTZ("Quartz");
- private final Integer code;
- private final String description;
+ private final String type;
- ScheduleStatus(Integer code, String description) {
- this.code = code;
- this.description = description;
+ ScheduleEngineType(String type) {
+ this.type = type;
}
-
- public static ScheduleStatus forCode(int code) {
- for (ScheduleStatus status : values()) {
- if (status.getCode() == code) {
- return status;
- }
- }
- throw new IllegalStateException(String.format("Illegal code=%s for ScheduleStatus", code));
- }
-
-}
+}
\ No newline at end of file
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleClient.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleClient.java
index 05aa6c01ae..6b7afe784f 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleClient.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleClient.java
@@ -19,18 +19,25 @@ package org.apache.inlong.manager.schedule.quartz;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
import org.apache.inlong.manager.schedule.ScheduleEngineClient;
+import org.apache.inlong.manager.schedule.ScheduleEngineType;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
/**
* Built-in implementation of schedule engine client corresponding with {@link QuartzScheduleEngine}.
* QuartzScheduleClient simply invokes the {@link QuartzScheduleEngine} to register/unregister/update
* schedule info instead of calling a remote schedule service.
* */
+@Service
public class QuartzScheduleClient implements ScheduleEngineClient {
- private final QuartzScheduleEngine scheduleEngine;
+ @Autowired
+ public QuartzScheduleEngine scheduleEngine;
- public QuartzScheduleClient(QuartzScheduleEngine scheduleEngine) {
- this.scheduleEngine = scheduleEngine;
+ @Override
+ public boolean accept(String engineType) {
+ return ScheduleEngineType.QUARTZ.getType().equalsIgnoreCase(engineType);
}
@Override
@@ -39,8 +46,8 @@ public class QuartzScheduleClient implements ScheduleEngineClient {
}
@Override
- public boolean unregister(ScheduleInfo scheduleInfo) {
- return scheduleEngine.handleUnregister(scheduleInfo);
+ public boolean unregister(String groupId) {
+ return scheduleEngine.handleUnregister(groupId);
}
@Override
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngine.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngine.java
index d9d2620211..e8bb1085ce 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngine.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngine.java
@@ -23,6 +23,7 @@ import org.apache.inlong.manager.schedule.exception.QuartzScheduleException;
import com.google.common.annotations.VisibleForTesting;
import lombok.Getter;
+import org.quartz.Job;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
@@ -31,6 +32,7 @@ import org.quartz.Trigger;
import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
import java.util.HashSet;
import java.util.Set;
@@ -43,6 +45,7 @@ import static org.apache.inlong.manager.schedule.util.ScheduleUtils.genQuartzTri
* the register/unregister/update requests from {@link QuartzScheduleClient}
* */
@Getter
+@Service
public class QuartzScheduleEngine implements ScheduleEngine {
private static final Logger LOGGER = LoggerFactory.getLogger(QuartzScheduleEngine.class);
@@ -90,7 +93,7 @@ public class QuartzScheduleEngine implements ScheduleEngine {
}
@VisibleForTesting
- public boolean handleRegister(ScheduleInfo scheduleInfo, Class<? extends QuartzOfflineSyncJob> clz) {
+ public boolean handleRegister(ScheduleInfo scheduleInfo, Class<? extends Job> clz) {
if (scheduledJobSet.contains(scheduleInfo.getInlongGroupId())) {
throw new QuartzScheduleException("Group " + scheduleInfo.getInlongGroupId() + " is already registered");
}
@@ -108,19 +111,19 @@ public class QuartzScheduleEngine implements ScheduleEngine {
/**
* Handle schedule unregister.
- * @param scheduleInfo schedule info to unregister
+ * @param groupId group to un-register schedule info
* */
@Override
- public boolean handleUnregister(ScheduleInfo scheduleInfo) {
- if (scheduledJobSet.contains(scheduleInfo.getInlongGroupId())) {
+ public boolean handleUnregister(String groupId) {
+ if (scheduledJobSet.contains(groupId)) {
try {
- scheduler.deleteJob(new JobKey(scheduleInfo.getInlongGroupId()));
+ scheduler.deleteJob(new JobKey(groupId));
} catch (SchedulerException e) {
throw new QuartzScheduleException(e.getMessage());
}
}
- scheduledJobSet.remove(scheduleInfo.getInlongGroupId());
- LOGGER.info("Un-registered schedule info for {}", scheduleInfo.getInlongGroupId());
+ scheduledJobSet.remove(groupId);
+ LOGGER.info("Un-registered schedule info for {}", groupId);
return true;
}
@@ -134,8 +137,8 @@ public class QuartzScheduleEngine implements ScheduleEngine {
}
@VisibleForTesting
- public boolean handleUpdate(ScheduleInfo scheduleInfo, Class<? extends QuartzOfflineSyncJob> clz) {
- handleUnregister(scheduleInfo);
+ public boolean handleUpdate(ScheduleInfo scheduleInfo, Class<? extends Job> clz) {
+ handleUnregister(scheduleInfo.getInlongGroupId());
handleRegister(scheduleInfo, clz);
LOGGER.info("Updated schedule info for {}", scheduleInfo.getInlongGroupId());
return false;
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/util/ScheduleUtils.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/util/ScheduleUtils.java
index 65475a0b98..1e4f43983e 100644
--- a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/util/ScheduleUtils.java
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/manager/schedule/util/ScheduleUtils.java
@@ -21,11 +21,11 @@ import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
import org.apache.inlong.manager.schedule.ScheduleType;
import org.apache.inlong.manager.schedule.ScheduleUnit;
import org.apache.inlong.manager.schedule.exception.QuartzScheduleException;
-import org.apache.inlong.manager.schedule.quartz.QuartzOfflineSyncJob;
import org.apache.commons.lang3.StringUtils;
import org.quartz.CronScheduleBuilder;
import org.quartz.CronTrigger;
+import org.quartz.Job;
import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.ScheduleBuilder;
@@ -42,7 +42,7 @@ import java.util.Date;
* */
public class ScheduleUtils {
- public static JobDetail genQuartzJobDetail(ScheduleInfo scheduleInfo, Class<? extends QuartzOfflineSyncJob> clz) {
+ public static JobDetail genQuartzJobDetail(ScheduleInfo scheduleInfo, Class<? extends Job> clz) {
return JobBuilder.newJob(clz)
.withIdentity(scheduleInfo.getInlongGroupId())
.build();
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/MockJob.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/MockJob.java
index 9202ea5b40..bc0fecf99a 100644
--- a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/MockJob.java
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/MockJob.java
@@ -17,6 +17,7 @@
package org.apache.inlong.manager.schedule.quartz;
+import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
@@ -25,7 +26,7 @@ import org.slf4j.LoggerFactory;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
-public class MockJob extends QuartzOfflineSyncJob {
+public class MockJob implements Job {
private static final Logger LOGGER = LoggerFactory.getLogger(MockJob.class);
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngineTest.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngineTest.java
index 008a7e42f8..11e7580f28 100644
--- a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngineTest.java
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/quartz/QuartzScheduleEngineTest.java
@@ -103,7 +103,7 @@ public class QuartzScheduleEngineTest extends BaseScheduleTest {
MockJob.countDownLatch.await();
// un-register before trigger finalized
- scheduleEngine.handleUnregister(scheduleInfo);
+ scheduleEngine.handleUnregister(scheduleInfo.getInlongGroupId());
// not job exist after un-register
assertEquals(0, scheduleEngine.getScheduledJobSet().size());
exist = scheduleEngine.getScheduler().checkExists(jobKey);
diff --git a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/util/ScheduleUtilsTest.java b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/util/ScheduleUtilsTest.java
index 415331be3d..da8fd66c61 100644
--- a/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/util/ScheduleUtilsTest.java
+++ b/inlong-manager/manager-schedule/src/test/java/org/apache/inlong/manager/schedule/util/ScheduleUtilsTest.java
@@ -20,7 +20,7 @@ package org.apache.inlong.manager.schedule.util;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
import org.apache.inlong.manager.schedule.BaseScheduleTest;
import org.apache.inlong.manager.schedule.exception.QuartzScheduleException;
-import org.apache.inlong.manager.schedule.quartz.QuartzOfflineSyncJob;
+import org.apache.inlong.manager.schedule.quartz.MockJob;
import org.junit.jupiter.api.Test;
import org.quartz.CronScheduleBuilder;
@@ -102,7 +102,7 @@ public class ScheduleUtilsTest extends BaseScheduleTest {
@Test
public void testGenJobDetail() {
ScheduleInfo scheduleInfo = genDefaultScheduleInfo();
- JobDetail jobDetail = ScheduleUtils.genQuartzJobDetail(scheduleInfo, QuartzOfflineSyncJob.class);
+ JobDetail jobDetail = ScheduleUtils.genQuartzJobDetail(scheduleInfo, MockJob.class);
assertNotNull(jobDetail);
JobKey jobKey = jobDetail.getKey();
@@ -116,7 +116,7 @@ public class ScheduleUtilsTest extends BaseScheduleTest {
public void testGenCronTrigger() {
// normal
ScheduleInfo scheduleInfo = genDefaultScheduleInfo();
- JobDetail jobDetail = ScheduleUtils.genQuartzJobDetail(scheduleInfo, QuartzOfflineSyncJob.class);
+ JobDetail jobDetail = ScheduleUtils.genQuartzJobDetail(scheduleInfo, MockJob.class);
Trigger trigger = ScheduleUtils.genQuartzTrigger(jobDetail, scheduleInfo);
assertNotNull(trigger);
@@ -139,7 +139,7 @@ public class ScheduleUtilsTest extends BaseScheduleTest {
// cron
scheduleInfo = genDefaultCronScheduleInfo();
- jobDetail = ScheduleUtils.genQuartzJobDetail(scheduleInfo, QuartzOfflineSyncJob.class);
+ jobDetail = ScheduleUtils.genQuartzJobDetail(scheduleInfo, MockJob.class);
trigger = ScheduleUtils.genQuartzTrigger(jobDetail, scheduleInfo);
assertNotNull(trigger);
diff --git a/inlong-manager/manager-service/pom.xml b/inlong-manager/manager-service/pom.xml
index 1b300dfc34..ae42b902b7 100644
--- a/inlong-manager/manager-service/pom.xml
+++ b/inlong-manager/manager-service/pom.xml
@@ -52,6 +52,11 @@
<artifactId>manager-workflow</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>manager-schedule</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>manager-test</artifactId>
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
index c9f4019a4d..d42a61dbfb 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
@@ -59,6 +59,8 @@ import org.apache.inlong.manager.pojo.group.InlongGroupPageRequest;
import org.apache.inlong.manager.pojo.group.InlongGroupRequest;
import org.apache.inlong.manager.pojo.group.InlongGroupTopicInfo;
import org.apache.inlong.manager.pojo.group.InlongGroupTopicRequest;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
import org.apache.inlong.manager.pojo.sink.StreamSink;
import org.apache.inlong.manager.pojo.sort.BaseSortConf;
import org.apache.inlong.manager.pojo.sort.BaseSortConf.SortType;
@@ -72,6 +74,7 @@ import org.apache.inlong.manager.pojo.user.LoginUserUtils;
import org.apache.inlong.manager.pojo.user.UserInfo;
import org.apache.inlong.manager.pojo.workflow.form.process.GroupResourceProcessForm;
import org.apache.inlong.manager.service.cluster.InlongClusterService;
+import org.apache.inlong.manager.service.schedule.ScheduleOperator;
import org.apache.inlong.manager.service.sink.StreamSinkService;
import org.apache.inlong.manager.service.source.SourceOperatorFactory;
import org.apache.inlong.manager.service.source.StreamSourceOperator;
@@ -111,6 +114,7 @@ import static org.apache.inlong.common.constant.ClusterSwitch.BACKUP_CLUSTER_TAG
import static org.apache.inlong.common.constant.ClusterSwitch.BACKUP_MQ_RESOURCE;
import static org.apache.inlong.common.constant.ClusterSwitch.CLUSTER_SWITCH_TIME;
import static org.apache.inlong.common.constant.ClusterSwitch.FINISH_SWITCH_INTERVAL_MIN;
+import static org.apache.inlong.manager.common.consts.InlongConstants.DATASYNC_OFFLINE_MODE;
import static org.apache.inlong.manager.pojo.common.PageRequest.MAX_PAGE_SIZE;
import static org.apache.inlong.manager.workflow.event.process.ProcessEventListener.EXECUTOR_SERVICE;
@@ -158,6 +162,9 @@ public class InlongGroupServiceImpl implements InlongGroupService {
@Autowired
private TenantUserRoleEntityMapper tenantUserRoleEntityMapper;
+ @Autowired
+ ScheduleOperator scheduleOperator;
+
/**
* Check whether modification is supported under the current group status, and which fields can be modified.
*
@@ -208,6 +215,11 @@ public class InlongGroupServiceImpl implements InlongGroupService {
// save ext info
this.saveOrUpdateExt(groupId, request.getExtList());
+ // save schedule info for offline group
+ if (DATASYNC_OFFLINE_MODE.equals(request.getInlongGroupMode())) {
+ scheduleOperator.saveOpt(CommonBeanUtils.copyProperties(request, ScheduleInfoRequest::new), operator);
+ }
+
LOGGER.info("success to save inlong group for groupId={} by user={}", groupId, operator);
return groupId;
}
@@ -239,7 +251,15 @@ public class InlongGroupServiceImpl implements InlongGroupService {
Preconditions.expectNotNull(groupId, ErrorCodeEnum.GROUP_ID_IS_EMPTY.getMessage());
InlongGroupEntity entity = groupMapper.selectByGroupId(groupId);
LOGGER.debug("success to check inlong group {}, exist? {}", groupId, entity != null);
- return entity != null;
+ if (entity == null) {
+ return false;
+ }
+ return isScheduleInfoExist(entity);
+ }
+
+ private boolean isScheduleInfoExist(InlongGroupEntity entity) {
+ return DATASYNC_OFFLINE_MODE.equals(entity.getInlongGroupMode())
+ && scheduleOperator.scheduleInfoExist(entity.getInlongGroupId());
}
@Override
@@ -261,11 +281,29 @@ public class InlongGroupServiceImpl implements InlongGroupService {
List<InlongStreamExtEntity> streamExtEntities = streamExtMapper.selectByRelatedId(groupId, null);
BaseSortConf sortConf = buildSortConfig(streamExtEntities);
groupInfo.setSortConf(sortConf);
-
+ if (DATASYNC_OFFLINE_MODE.equals(entity.getInlongGroupMode())) {
+ // get schedule info and set into group info
+ addScheduleInfo(entity, groupInfo);
+ }
LOGGER.debug("success to get inlong group for groupId={}", groupId);
return groupInfo;
}
+ private void addScheduleInfo(InlongGroupEntity entity, InlongGroupInfo groupInfo) {
+ checkOfflineSyncScheduleExist(entity);
+ ScheduleInfo scheduleInfo = scheduleOperator.getScheduleInfo(entity.getInlongGroupId());
+ CommonBeanUtils.copyProperties(scheduleInfo, groupInfo);
+ }
+
+ private void checkOfflineSyncScheduleExist(InlongGroupEntity entity) {
+ // check schedule info for offline sync
+ if (!isScheduleInfoExist(entity)) {
+ String errorMsg = String.format("Schedule info not found for groupId=%s", entity.getInlongGroupId());
+ LOGGER.error(errorMsg);
+ throw new BusinessException(ErrorCodeEnum.SCHEDULE_NOT_FOUND, errorMsg);
+ }
+ }
+
@Override
public String getTenant(String groupId, String operator) {
InlongGroupEntity groupEntity = groupMapper.selectByGroupIdWithoutTenant(groupId);
@@ -457,6 +495,12 @@ public class InlongGroupServiceImpl implements InlongGroupService {
// save ext info
this.saveOrUpdateExt(groupId, request.getExtList());
+ // save schedule info for offline group
+ if (DATASYNC_OFFLINE_MODE.equals(request.getInlongGroupMode())) {
+ scheduleOperator.updateAndRegister(CommonBeanUtils.copyProperties(request, ScheduleInfoRequest::new),
+ operator);
+ }
+
LOGGER.info("success to update inlong group for groupId={} by user={}", groupId, operator);
return groupId;
}
@@ -612,6 +656,15 @@ public class InlongGroupServiceImpl implements InlongGroupService {
// logically delete the associated extension info
groupExtMapper.logicDeleteAllByGroupId(groupId);
+ // remove schedule
+ if (DATASYNC_OFFLINE_MODE.equals(entity.getInlongGroupMode())) {
+ try {
+ scheduleOperator.deleteByGroupIdOpt(entity.getInlongGroupId(), operator);
+ } catch (Exception e) {
+ LOGGER.warn("failed to delete schedule info for groupId={}, error msg: {}", groupId, e.getMessage());
+ }
+ }
+
LOGGER.info("success to delete group and group ext property for groupId={} by user={}", groupId, operator);
return true;
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/GroupScheduleResourceListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/GroupScheduleResourceListener.java
index e27dab9e89..39d3a2a3bf 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/GroupScheduleResourceListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/schedule/GroupScheduleResourceListener.java
@@ -24,11 +24,13 @@ import org.apache.inlong.manager.pojo.group.InlongGroupExtInfo;
import org.apache.inlong.manager.pojo.group.InlongGroupInfo;
import org.apache.inlong.manager.pojo.workflow.form.process.GroupResourceProcessForm;
import org.apache.inlong.manager.pojo.workflow.form.process.ProcessForm;
+import org.apache.inlong.manager.service.schedule.ScheduleOperator;
import org.apache.inlong.manager.workflow.WorkflowContext;
import org.apache.inlong.manager.workflow.event.ListenerResult;
import org.apache.inlong.manager.workflow.event.task.ScheduleOperateListener;
import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.List;
@@ -37,6 +39,9 @@ import java.util.List;
@Slf4j
public class GroupScheduleResourceListener implements ScheduleOperateListener {
+ @Autowired
+ private ScheduleOperator scheduleOperator;
+
@Override
public TaskEvent event() {
return TaskEvent.COMPLETE;
@@ -68,7 +73,8 @@ public class GroupScheduleResourceListener implements ScheduleOperateListener {
final String groupId = groupInfo.getInlongGroupId();
log.info("begin to register schedule info for groupId={}", groupId);
- // todo: register schedule info to schedule service
+ // handle schedule info after group approved
+ scheduleOperator.handleGroupApprove(groupId);
// after register schedule info successfully, add ext property to group ext info
saveInfo(groupInfo, InlongConstants.REGISTER_SCHEDULE_STATUS,
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleOperator.java
new file mode 100644
index 0000000000..6bf9c01432
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleOperator.java
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.service.schedule;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
+
+/**
+ * Operator for schedule. Including:
+ * 1. schedule info management
+ * 2. schedule operations like resister, un-register etc.
+ * */
+public interface ScheduleOperator {
+
+ /**
+ * Save schedule info.
+ * There are two places may save schedule info:
+ * - 1. create new inlong group with schedule info
+ * - 2. create new schedule info directly(inlong group has been already exist), in this situation, we should
+ * register schedule info to schedule engine if group has been approved.
+ * @param request schedule request need to save
+ * @param operator name of operator
+ * @return schedule info id in backend storage
+ */
+ int saveOpt(ScheduleInfoRequest request, String operator);
+
+ /**
+ * Check whether schedule info exists for specified inlong group
+ *
+ * @param groupId the group id to be queried
+ * @return does it exist
+ */
+ Boolean scheduleInfoExist(String groupId);
+
+ /**
+ * Get schedule info based on inlong group id
+ *
+ * @param groupId inlong group id
+ * @return detail of inlong group
+ */
+ ScheduleInfo getScheduleInfo(String groupId);
+
+ /**
+ * Modify schedule information
+ * There are two places may update schedule info:
+ * - 1. update inlong group with new schedule info
+ * - 2. update schedule info directly(inlong group has been already exist)
+ * @param request schedule request that needs to be modified
+ * @param operator name of operator
+ * @return whether succeed
+ */
+ Boolean updateOpt(ScheduleInfoRequest request, String operator);
+
+ /**
+ * Register schedule information
+ * @param request schedule request that needs to be modified
+ * @param operator name of operator
+ * @return whether succeed
+ */
+ Boolean updateAndRegister(ScheduleInfoRequest request, String operator);
+
+ /**
+ * Delete schedule info for groupId.
+ * There are two places may delete schedule info:
+ * - 1. delete an inlong group
+ * - 2. delete schedule info directly, left inlong group alone without schedule info, which means the group of
+ * offline sync job will never be triggered
+ * @param groupId groupId to find a schedule info to delete
+ * @param operator name of operator
+ * @Return whether succeed
+ * */
+ Boolean deleteByGroupIdOpt(String groupId, String operator);
+
+ /**
+ * Handle inlong group approve, check schedule info and try to register it to schedule engine.
+ * @param groupId groupId to find a schedule info to delete
+ * @Return whether succeed
+ * */
+ Boolean handleGroupApprove(String groupId);
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleOperatorImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleOperatorImpl.java
new file mode 100644
index 0000000000..411a80766a
--- /dev/null
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleOperatorImpl.java
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.manager.service.schedule;
+
+import org.apache.inlong.manager.common.consts.InlongConstants;
+import org.apache.inlong.manager.common.util.CommonBeanUtils;
+import org.apache.inlong.manager.dao.entity.InlongGroupExtEntity;
+import org.apache.inlong.manager.dao.mapper.InlongGroupExtEntityMapper;
+import org.apache.inlong.manager.dao.mapper.ScheduleEntityMapper;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
+import org.apache.inlong.manager.schedule.ScheduleClientFactory;
+import org.apache.inlong.manager.schedule.ScheduleEngineClient;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import static org.apache.inlong.manager.common.enums.ScheduleStatus.APPROVED;
+import static org.apache.inlong.manager.common.enums.ScheduleStatus.REGISTERED;
+import static org.apache.inlong.manager.common.enums.ScheduleStatus.UPDATED;
+
+@Service
+public class ScheduleOperatorImpl implements ScheduleOperator {
+
+ private static final Logger LOGGER = LoggerFactory.getLogger(ScheduleOperatorImpl.class);
+
+ @Autowired
+ private ScheduleService scheduleService;
+
+ @Autowired
+ private InlongGroupExtEntityMapper groupExtMapper;
+
+ @Autowired
+ private ScheduleEntityMapper scheduleMapper;
+
+ @Autowired
+ private ScheduleClientFactory scheduleClientFactory;
+
+ private ScheduleEngineClient scheduleEngineClient;
+
+ @Override
+ @Transactional(rollbackFor = Throwable.class)
+ public int saveOpt(ScheduleInfoRequest request, String operator) {
+ // save schedule info first
+ int scheduleInfoId = scheduleService.save(request, operator);
+ LOGGER.info("Save schedule info success for group {}", request.getInlongGroupId());
+ // process new schedule info for approved inlong group
+ registerScheduleInfoForApprovedGroup(CommonBeanUtils.copyProperties(request, ScheduleInfo::new), operator);
+ return scheduleInfoId;
+ }
+
+ /**
+ * If an inlong group in DATASYNC_OFFLINE_MODE created first without schedule info and has been approved, it should
+ * be registered to schedule engine once the schedule info for this group is added.
+ * */
+ private void registerScheduleInfoForApprovedGroup(ScheduleInfo scheduleInfo, String operator) {
+ String groupId = scheduleInfo.getInlongGroupId();
+ InlongGroupExtEntity scheduleStatusExt =
+ groupExtMapper.selectByUniqueKey(groupId, InlongConstants.REGISTER_SCHEDULE_STATUS);
+ if (InlongConstants.REGISTERED.equalsIgnoreCase(scheduleStatusExt.getKeyValue())) {
+ // change schedule state to approved
+ scheduleService.updateStatus(scheduleInfo.getInlongGroupId(), APPROVED, operator);
+ registerToScheduleEngine(scheduleInfo, operator, false);
+ LOGGER.info("Register schedule info success for group {}", groupId);
+ }
+ }
+
+ private ScheduleEngineClient getScheduleEngineClient() {
+ if (scheduleEngineClient == null) {
+ scheduleEngineClient = scheduleClientFactory.getInstance();
+ }
+ return scheduleEngineClient;
+ }
+
+ @Override
+ public Boolean scheduleInfoExist(String groupId) {
+ return scheduleService.exist(groupId);
+ }
+
+ @Override
+ public ScheduleInfo getScheduleInfo(String groupId) {
+ return scheduleService.get(groupId);
+ }
+
+ @Override
+ @Transactional(rollbackFor = Throwable.class)
+ public Boolean updateOpt(ScheduleInfoRequest request, String operator) {
+ // if the inlong group exist without schedule info
+ // then, save the new schedule info when updating inlong group
+ if (!scheduleInfoExist(request.getInlongGroupId())) {
+ saveOpt(request, operator);
+ return true;
+ }
+ ScheduleInfo scheduleInfo = CommonBeanUtils.copyProperties(request, ScheduleInfo::new);
+ if (!needUpdate(scheduleInfo)) {
+ LOGGER.info("schedule info not changed for group {}", request.getInlongGroupId());
+ return false;
+ }
+ // update schedule info
+ boolean res = scheduleService.update(request, operator);
+ // update status
+ scheduleService.updateStatus(request.getInlongGroupId(), UPDATED, operator);
+ return res;
+ }
+
+ @Override
+ @Transactional(rollbackFor = Throwable.class)
+ public Boolean updateAndRegister(ScheduleInfoRequest request, String operator) {
+ updateOpt(request, operator);
+ return registerToScheduleEngine(CommonBeanUtils.copyProperties(request, ScheduleInfo::new), operator, true);
+ }
+
+ /**
+ * There are three places may trigger resister schedule info to schedule engine:
+ * - 1. new group approved with schedule info
+ * - 2. new schedule info for an exist approved inlong group added
+ * - 3. group's schedule info updated
+ * */
+ private Boolean registerToScheduleEngine(ScheduleInfo scheduleInfo, String operator, boolean isUpdate) {
+ // update(un-register and then register) or register
+ boolean res = isUpdate ? getScheduleEngineClient().update(scheduleInfo)
+ : getScheduleEngineClient().register(scheduleInfo);
+ // update status to REGISTERED
+ scheduleService.updateStatus(scheduleInfo.getInlongGroupId(), REGISTERED, operator);
+ LOGGER.info("{} schedule info success for group {}",
+ isUpdate ? "Update" : "Register", scheduleInfo.getInlongGroupId());
+ return res;
+ }
+
+ private boolean needUpdate(ScheduleInfo scheduleInfo) {
+ if (scheduleInfo == null) {
+ return false;
+ }
+ ScheduleInfo existedSchedule = getScheduleInfo(scheduleInfo.getInlongGroupId());
+ return !scheduleInfo.equals(existedSchedule);
+ }
+
+ @Override
+ public Boolean deleteByGroupIdOpt(String groupId, String operator) {
+ return scheduleService.deleteByGroupId(groupId, operator);
+ }
+
+ @Override
+ public Boolean handleGroupApprove(String groupId) {
+ // if the inlong group exist without schedule info
+ // then, save the new schedule info when updating inlong group
+ if (!scheduleInfoExist(groupId)) {
+ LOGGER.warn("schedule info not exist for group {}", groupId);
+ return false;
+ }
+ ScheduleInfo scheduleInfo = getScheduleInfo(groupId);
+ // change schedule state to approved
+ scheduleService.updateStatus(groupId, APPROVED, null);
+ return registerToScheduleEngine(scheduleInfo, null, false);
+ }
+
+}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleService.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleService.java
index d00e7134d4..f3eb189c38 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleService.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleService.java
@@ -17,6 +17,7 @@
package org.apache.inlong.manager.service.schedule;
+import org.apache.inlong.manager.common.enums.ScheduleStatus;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
@@ -61,6 +62,16 @@ public interface ScheduleService {
Boolean update(@Valid @NotNull(message = "schedule request cannot be null") ScheduleInfoRequest request,
String operator);
+ /**
+ * Update status of schedule info.
+ *
+ * @param groupId group to update schedule status
+ * @param newStatus status to update
+ * @param operator name of operator
+ * @return whether succeed
+ */
+ Boolean updateStatus(String groupId, ScheduleStatus newStatus, String operator);
+
/**
* Delete schedule info for gropuId.
* @param groupId groupId to find a schedule info to delete
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleServiceImpl.java
index 480189da9e..6459a811cd 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/schedule/ScheduleServiceImpl.java
@@ -35,7 +35,18 @@ import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
import java.util.Objects;
+import java.util.Set;
+
+import static org.apache.inlong.manager.common.enums.ScheduleStatus.APPROVED;
+import static org.apache.inlong.manager.common.enums.ScheduleStatus.DELETED;
+import static org.apache.inlong.manager.common.enums.ScheduleStatus.NEW;
+import static org.apache.inlong.manager.common.enums.ScheduleStatus.REGISTERED;
+import static org.apache.inlong.manager.common.enums.ScheduleStatus.UPDATED;
@Service
public class ScheduleServiceImpl implements ScheduleService {
@@ -47,6 +58,9 @@ public class ScheduleServiceImpl implements ScheduleService {
@Autowired
private ScheduleEntityMapper scheduleEntityMapper;
+ // finite state machine
+ private Map<ScheduleStatus, Set<ScheduleStatus>> fsm;
+
@Override
public int save(ScheduleInfoRequest request, String operator) {
LOGGER.debug("begin to save schedule info, scheduleInfo={}, operator={}", request, operator);
@@ -62,7 +76,8 @@ public class ScheduleServiceImpl implements ScheduleService {
scheduleEntity.setStatus(ScheduleStatus.NEW.getCode());
scheduleEntity.setCreator(operator);
scheduleEntity.setModifier(operator);
- return scheduleEntityMapper.insert(scheduleEntity);
+ scheduleEntityMapper.insert(scheduleEntity);
+ return scheduleEntity.getId();
}
@Override
@@ -97,6 +112,43 @@ public class ScheduleServiceImpl implements ScheduleService {
return true;
}
+ @Override
+ public Boolean updateStatus(String groupId, ScheduleStatus newStatus, String operator) {
+ LOGGER.debug("begin to update schedule status for groupId={}", groupId);
+ ScheduleEntity entity = getScheduleEntity(groupId);
+ ScheduleStatus preStatus = ScheduleStatus.forCode(entity.getStatus());
+ if (!isAllowedTransition(preStatus, newStatus)) {
+ String errorMsg = String.format("Schedule status transition is not allowed from %s to %s for group %s",
+ preStatus, newStatus, groupId);
+ LOGGER.error(errorMsg);
+ throw new BusinessException(ErrorCodeEnum.SCHEDULE_STATUS_TRANSITION_NOT_ALLOWED);
+ }
+ entity.setStatus(newStatus.getCode());
+ entity.setModifier(operator);
+ updateScheduleInfo(entity,
+ String.format("update schedule status from %s to %s failed for groupId=%s",
+ preStatus.getCode(), newStatus.getCode(), entity.getInlongGroupId()));
+ LOGGER.info("success to update schedule status from {} to {} for groupId={}",
+ preStatus.getCode(), newStatus.getCode(), groupId);
+ return true;
+ }
+
+ private void initFSMIfNeed() {
+ if (fsm != null) {
+ return;
+ }
+ fsm = new HashMap<>();
+ fsm.put(NEW, new HashSet<>(Arrays.asList(APPROVED, DELETED)));
+ fsm.put(APPROVED, new HashSet<>(Arrays.asList(REGISTERED, DELETED)));
+ fsm.put(REGISTERED, new HashSet<>(Arrays.asList(UPDATED, DELETED)));
+ fsm.put(UPDATED, new HashSet<>(Arrays.asList(REGISTERED, DELETED)));
+ }
+
+ private boolean isAllowedTransition(ScheduleStatus preStatus, ScheduleStatus newStatus) {
+ initFSMIfNeed();
+ return fsm.get(preStatus).contains(newStatus);
+ }
+
@Override
public Boolean deleteByGroupId(String groupId, String operator) {
LOGGER.debug("begin to delete schedule info for groupId={}", groupId);
diff --git a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InLongSchedulerController.java b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InLongSchedulerController.java
index 801cc09b0e..4af07854c2 100644
--- a/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InLongSchedulerController.java
+++ b/inlong-manager/manager-web/src/main/java/org/apache/inlong/manager/web/controller/InLongSchedulerController.java
@@ -25,7 +25,7 @@ import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
import org.apache.inlong.manager.pojo.schedule.ScheduleInfoRequest;
import org.apache.inlong.manager.pojo.user.LoginUserUtils;
import org.apache.inlong.manager.service.operationlog.OperationLog;
-import org.apache.inlong.manager.service.schedule.ScheduleService;
+import org.apache.inlong.manager.service.schedule.ScheduleOperator;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
@@ -45,35 +45,43 @@ import org.springframework.web.bind.annotation.RestController;
public class InLongSchedulerController {
@Autowired
- private ScheduleService scheduleService;
+ private ScheduleOperator scheduleOperator;
@RequestMapping(value = "/schedule/save", method = RequestMethod.POST)
@OperationLog(operation = OperationType.CREATE, operationTarget = OperationTarget.SCHEDULE)
@ApiOperation(value = "Save schedule info")
public Response<Integer> save(@RequestBody ScheduleInfoRequest request) {
- int result = scheduleService.save(request, LoginUserUtils.getLoginUser().getName());
- return Response.success(result);
+ int scheduleInfoId = scheduleOperator.saveOpt(request, LoginUserUtils.getLoginUser().getName());
+ return Response.success(scheduleInfoId);
}
@RequestMapping(value = "/schedule/exist/{groupId}", method = RequestMethod.GET)
@ApiOperation(value = "Is the schedule info exists for inlong group")
@ApiImplicitParam(name = "groupId", dataTypeClass = String.class, required = true)
public Response<Boolean> exist(@PathVariable String groupId) {
- return Response.success(scheduleService.exist(groupId));
+ return Response.success(scheduleOperator.scheduleInfoExist(groupId));
}
@RequestMapping(value = "/schedule/get", method = RequestMethod.GET)
@ApiOperation(value = "Get schedule info for inlong group")
@ApiImplicitParam(name = "groupId", dataTypeClass = String.class, required = true)
public Response<ScheduleInfo> get(@RequestParam String groupId) {
- return Response.success(scheduleService.get(groupId));
+ return Response.success(scheduleOperator.getScheduleInfo(groupId));
}
@RequestMapping(value = "/schedule/update", method = RequestMethod.POST)
@OperationLog(operation = OperationType.UPDATE, operationTarget = OperationTarget.SCHEDULE)
@ApiOperation(value = "Update schedule info")
public Response<Boolean> update(@Validated(UpdateValidation.class) @RequestBody ScheduleInfoRequest request) {
- return Response.success(scheduleService.update(request, LoginUserUtils.getLoginUser().getName()));
+ return Response.success(scheduleOperator.updateOpt(request, LoginUserUtils.getLoginUser().getName()));
+ }
+
+ @RequestMapping(value = "/schedule/updateAndRegister", method = RequestMethod.POST)
+ @OperationLog(operation = OperationType.UPDATE, operationTarget = OperationTarget.SCHEDULE)
+ @ApiOperation(value = "Update schedule info and register to schedule engine")
+ public Response<Boolean> updateAndRegister(
+ @Validated(UpdateValidation.class) @RequestBody ScheduleInfoRequest request) {
+ return Response.success(scheduleOperator.updateAndRegister(request, LoginUserUtils.getLoginUser().getName()));
}
@RequestMapping(value = "/schedule/delete/{groupId}", method = RequestMethod.DELETE)
@@ -82,7 +90,7 @@ public class InLongSchedulerController {
@ApiImplicitParam(name = "groupId", value = "Inlong group id", dataTypeClass = String.class, required = true)
public Response<Boolean> delete(@PathVariable String groupId) {
String operator = LoginUserUtils.getLoginUser().getName();
- return Response.success(scheduleService.deleteByGroupId(groupId, operator));
+ return Response.success(scheduleOperator.deleteByGroupIdOpt(groupId, operator));
}
}
diff --git a/inlong-manager/manager-web/src/main/resources/application-dev.properties b/inlong-manager/manager-web/src/main/resources/application-dev.properties
index 81b32bb941..60f82953a5 100644
--- a/inlong-manager/manager-web/src/main/resources/application-dev.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-dev.properties
@@ -98,4 +98,8 @@ cls.manager.endpoint=127.0.0.1
manager.url=127.0.0.1:8083
-agent.install.path=
\ No newline at end of file
+agent.install.path=
+
+# schedule engine type
+# support none(no scheduler) and quartz(quartz scheduler), default is none
+inlong.schedule.engine=none
\ No newline at end of file
diff --git a/inlong-manager/manager-web/src/main/resources/application-prod.properties b/inlong-manager/manager-web/src/main/resources/application-prod.properties
index 6143155bf0..3e8f329470 100644
--- a/inlong-manager/manager-web/src/main/resources/application-prod.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-prod.properties
@@ -95,3 +95,6 @@ group.deleted.enabled=false
# Tencent cloud log service endpoint, The Operator cls resource by it
cls.manager.endpoint=127.0.0.1
+# schedule engine type
+# support none(no scheduler) and quartz(quartz scheduler), default is none
+inlong.schedule.engine=none
\ No newline at end of file
diff --git a/inlong-manager/manager-web/src/main/resources/application-test.properties b/inlong-manager/manager-web/src/main/resources/application-test.properties
index dcab3fb1cf..5ff929c2b8 100644
--- a/inlong-manager/manager-web/src/main/resources/application-test.properties
+++ b/inlong-manager/manager-web/src/main/resources/application-test.properties
@@ -96,3 +96,6 @@ group.deleted.enabled=false
# Tencent cloud log service endpoint, The Operator cls resource by it
cls.manager.endpoint=127.0.0.1
+# schedule engine type
+# support none(no scheduler) and quartz(quartz scheduler), default is none
+inlong.schedule.engine=none
\ No newline at end of file
diff --git a/inlong-manager/manager-web/src/main/resources/application.properties b/inlong-manager/manager-web/src/main/resources/application.properties
index 6b56dfb3d9..a6eec820ad 100644
--- a/inlong-manager/manager-web/src/main/resources/application.properties
+++ b/inlong-manager/manager-web/src/main/resources/application.properties
@@ -66,3 +66,7 @@ audit.user.ids=3,4,5,6
# tencent cloud log service endpoint, The Operator cls resource by it
cls.manager.endpoint=127.0.0.1
+
+# schedule engine type
+# support none(no scheduler) and quartz(quartz scheduler), default is none
+inlong.schedule.engine=none
\ No newline at end of file
(inlong) 02/09: [INLONG-9813][Manager] Support offline data sync management (#9814)
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 0bae684745f4d640689377c07f62adabfb7b1afd
Author: AloysZhang <al...@apache.org>
AuthorDate: Wed Mar 13 16:18:04 2024 +0800
[INLONG-9813][Manager] Support offline data sync management (#9814)
---
.../apache/inlong/manager/plugin/listener/StartupSortListener.java | 3 ++-
.../manager/service/listener/group/InitGroupCompleteListener.java | 3 ++-
.../manager/service/listener/group/UpdateGroupCompleteListener.java | 3 ++-
.../manager/service/listener/queue/QueueResourceListener.java | 6 ++++--
.../manager/service/listener/stream/InitStreamCompleteListener.java | 3 ++-
.../inlong/manager/service/source/AbstractSourceOperator.java | 1 +
.../inlong/manager/service/source/StreamSourceServiceImpl.java | 3 ++-
7 files changed, 15 insertions(+), 7 deletions(-)
diff --git a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
index 405cb92a3c..ae2de1c86f 100644
--- a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
+++ b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
@@ -61,7 +61,8 @@ public class StartupSortListener implements SortOperateListener {
}
log.info("add startup group listener for groupId [{}]", groupId);
- return InlongConstants.DATASYNC_REALTIME_MODE.equals(groupProcessForm.getGroupInfo().getInlongGroupMode());
+ return (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupProcessForm.getGroupInfo().getInlongGroupMode())
+ || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupProcessForm.getGroupInfo().getInlongGroupMode()));
}
@Override
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/InitGroupCompleteListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/InitGroupCompleteListener.java
index c96c12c9ff..c8c087320f 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/InitGroupCompleteListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/InitGroupCompleteListener.java
@@ -98,7 +98,8 @@ public class InitGroupCompleteListener implements ProcessEventListener {
// update status of other related configs
if (InlongConstants.DISABLE_CREATE_RESOURCE.equals(groupInfo.getEnableCreateResource())) {
- if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())
+ || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())) {
sourceService.updateStatus(groupId, null, SourceStatus.SOURCE_NORMAL.getCode(), operator);
} else {
sourceService.updateStatus(groupId, null, SourceStatus.TO_BE_ISSUED_ADD.getCode(), operator);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/UpdateGroupCompleteListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/UpdateGroupCompleteListener.java
index a6dd8fc786..1a6e1f7fe1 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/UpdateGroupCompleteListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/UpdateGroupCompleteListener.java
@@ -96,7 +96,8 @@ public class UpdateGroupCompleteListener implements ProcessEventListener {
}
// if the inlong group is dataSync mode, the stream source needs to be processed.
- if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())
+ || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())) {
changeSource4DataSync(groupId, operateType, operator);
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java
index f566c29728..0ce551d1e6 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java
@@ -115,8 +115,10 @@ public class QueueResourceListener implements QueueOperateListener {
String operator = context.getOperator();
GroupOperateType operateType = groupProcessForm.getGroupOperateType();
- if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
- log.warn("skip to execute QueueResourceListener as sync mode for groupId={}", groupId);
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())
+ || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())) {
+ log.warn("skip to execute QueueResourceListener as sync mode {} (1 for realtime sync, 2 for offline sync) "
+ + "for groupId={}", groupInfo.getInlongGroupMode(), groupId);
if (GroupOperateType.INIT.equals(operateType)) {
this.createQueueForStreams(groupInfo, groupProcessForm.getStreamInfos(), operator);
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/InitStreamCompleteListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/InitStreamCompleteListener.java
index df41823b84..85033b44f7 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/InitStreamCompleteListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/InitStreamCompleteListener.java
@@ -66,7 +66,8 @@ public class InitStreamCompleteListener implements ProcessEventListener {
// Update status of other related configs
streamService.updateStatus(groupId, streamId, StreamStatus.CONFIG_SUCCESSFUL.getCode(), operator);
streamService.updateWithoutCheck(streamInfo.genRequest(), operator);
- if (InlongConstants.DATASYNC_REALTIME_MODE.equals(form.getGroupInfo().getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(form.getGroupInfo().getInlongGroupMode())
+ || InlongConstants.DATASYNC_OFFLINE_MODE.equals(form.getGroupInfo().getInlongGroupMode())) {
sourceService.updateStatus(groupId, streamId, SourceStatus.SOURCE_NORMAL.getCode(), operator);
} else {
sourceService.updateStatus(groupId, streamId, SourceStatus.TO_BE_ISSUED_ADD.getCode(), operator);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
index 83b7ac3ea9..39d3845209 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
@@ -139,6 +139,7 @@ public abstract class AbstractSourceOperator implements StreamSourceOperator {
return;
}
boolean allowUpdate = InlongConstants.DATASYNC_REALTIME_MODE.equals(groupMode)
+ || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupMode)
|| SourceStatus.ALLOWED_UPDATE.contains(entity.getStatus());
if (!allowUpdate) {
throw new BusinessException(ErrorCodeEnum.SOURCE_OPT_NOT_ALLOWED,
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
index 1a9dc25243..aab60a8fdb 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
@@ -190,7 +190,8 @@ public class StreamSourceServiceImpl implements StreamSourceService {
// if the group mode is DATASYNC, just get all related stream sources
List<StreamSource> streamSources = this.listSource(groupId, null);
- if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())
+ || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupInfo.getInlongGroupMode())) {
result = streamSources.stream()
.collect(Collectors.groupingBy(StreamSource::getInlongStreamId, HashMap::new,
Collectors.toCollection(ArrayList::new)));
(inlong) 01/09: [INLONG-9781][Manager] Add offline sync task type definition (#9787)
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 565a67ba1706b95be969394c3933d6bda3c1d860
Author: AloysZhang <al...@apache.org>
AuthorDate: Thu Mar 7 11:44:20 2024 +0800
[INLONG-9781][Manager] Add offline sync task type definition (#9787)
---
.../java/org/apache/inlong/manager/client/ut/BaseTest.java | 2 +-
.../inlong/manager/common/consts/InlongConstants.java | 3 ++-
.../org/apache/inlong/manager/common/enums/GroupMode.java | 14 +++++++++++---
.../manager/plugin/listener/StartupSortListener.java | 2 +-
.../apache/inlong/manager/pojo/group/InlongGroupInfo.java | 3 ++-
.../inlong/manager/pojo/group/InlongGroupPageRequest.java | 3 ++-
.../inlong/manager/pojo/group/InlongGroupRequest.java | 5 +++--
.../inlong/manager/service/core/impl/AuditServiceImpl.java | 6 ++++--
.../manager/service/group/InlongGroupServiceImpl.java | 2 +-
.../service/listener/group/InitGroupCompleteListener.java | 2 +-
.../listener/group/UpdateGroupCompleteListener.java | 2 +-
.../service/listener/queue/QueueResourceListener.java | 2 +-
.../listener/stream/InitStreamCompleteListener.java | 2 +-
.../manager/service/source/AbstractSourceOperator.java | 2 +-
.../manager/service/source/StreamSourceServiceImpl.java | 2 +-
15 files changed, 33 insertions(+), 19 deletions(-)
diff --git a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/ut/BaseTest.java b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/ut/BaseTest.java
index 4d218918f5..9961c16ef8 100644
--- a/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/ut/BaseTest.java
+++ b/inlong-manager/manager-client-examples/src/test/java/org/apache/inlong/manager/client/ut/BaseTest.java
@@ -110,7 +110,7 @@ public class BaseTest {
// set enable zk, create resource, group mode, and cluster tag
pulsarInfo.setEnableZookeeper(InlongConstants.DISABLE_ZK);
pulsarInfo.setEnableCreateResource(InlongConstants.ENABLE_CREATE_RESOURCE);
- pulsarInfo.setInlongGroupMode(InlongConstants.DATASYNC_MODE);
+ pulsarInfo.setInlongGroupMode(InlongConstants.DATASYNC_REALTIME_MODE);
pulsarInfo.setInlongClusterTag("default_cluster");
pulsarInfo.setDailyRecords(10000000);
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java
index 820c22adcf..6ef9f5772c 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/consts/InlongConstants.java
@@ -87,7 +87,8 @@ public class InlongConstants {
public static final Integer DELETED_STATUS = 10;
public static final Integer STANDARD_MODE = 0;
- public static final Integer DATASYNC_MODE = 1;
+ public static final Integer DATASYNC_REALTIME_MODE = 1;
+ public static final Integer DATASYNC_OFFLINE_MODE = 2;
public static final Integer DISABLE_ZK = 0;
public static final Integer ENABLE_ZK = 1;
diff --git a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/GroupMode.java b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/GroupMode.java
index f0db2353b6..d4b417f39a 100644
--- a/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/GroupMode.java
+++ b/inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/enums/GroupMode.java
@@ -31,10 +31,18 @@ public enum GroupMode {
STANDARD("standard"),
/**
- * DataSync mode(only Data Synchronization): group init only with sort in InLong Cluster
- * StreamSource -> Sort -> StreamSink
+ * DataSync mode(only Data Synchronization): real-time data sync in stream way, group init only with
+ * sort in InLong Cluster.
+ * StreamSource -> Sort -> Sink
*/
- DATASYNC("datasync");
+ DATASYNC("datasync"),
+
+ /**
+ * DataSync mode(only Data Synchronization): offline data sync in batch way, group init only with sort
+ * in InLong Cluster.
+ * BatchSource -> Sort -> Sink
+ */
+ DATASYNC_BATCH("datasync_offline");
@Getter
private final String mode;
diff --git a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
index 001d967a56..405cb92a3c 100644
--- a/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
+++ b/inlong-manager/manager-plugins/base/src/main/java/org/apache/inlong/manager/plugin/listener/StartupSortListener.java
@@ -61,7 +61,7 @@ public class StartupSortListener implements SortOperateListener {
}
log.info("add startup group listener for groupId [{}]", groupId);
- return InlongConstants.DATASYNC_MODE.equals(groupProcessForm.getGroupInfo().getInlongGroupMode());
+ return InlongConstants.DATASYNC_REALTIME_MODE.equals(groupProcessForm.getGroupInfo().getInlongGroupMode());
}
@Override
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupInfo.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupInfo.java
index 5b3e730ed9..17b66497a9 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupInfo.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupInfo.java
@@ -72,7 +72,8 @@ public abstract class InlongGroupInfo extends BaseInlongGroup {
@ApiModelProperty(value = "Whether to enable create resource? 0: disable, 1: enable")
private Integer enableCreateResource;
- @ApiModelProperty(value = "Standard mode(include Data Ingestion and Synchronization): 0, DataSync mode(only Data Synchronization): 1")
+ @ApiModelProperty(value = "Standard mode(include Data Ingestion and Synchronization): 0, DataSync mode(only Data Synchronization, real-time data sync in stream way): 1,"
+ + " DataSync mode(only Data Synchronization, offline data sync in batch way): 2")
private Integer inlongGroupMode;
@ApiModelProperty(value = "Data report type, default is 0.\n"
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupPageRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupPageRequest.java
index 5d7c24b295..4dc2140166 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupPageRequest.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupPageRequest.java
@@ -58,7 +58,8 @@ public class InlongGroupPageRequest extends PageRequest {
@ApiModelProperty(value = "The inlong cluster tag list")
private List<String> clusterTagList;
- @ApiModelProperty(value = "Standard mode(include Data Ingestion and Synchronization): 0, DataSync mode(only Data Synchronization): 1")
+ @ApiModelProperty(value = "Standard mode(include Data Ingestion and Synchronization): 0, DataSync mode(only Data Synchronization, real-time data sync in stream way): 1,"
+ + " DataSync mode(only Data Synchronization, offline data sync in batch way): 2")
private Integer inlongGroupMode;
@ApiModelProperty(value = "Current user", hidden = true)
diff --git a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupRequest.java b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupRequest.java
index 30893b7a09..6140bddad5 100644
--- a/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupRequest.java
+++ b/inlong-manager/manager-pojo/src/main/java/org/apache/inlong/manager/pojo/group/InlongGroupRequest.java
@@ -86,8 +86,9 @@ public abstract class InlongGroupRequest extends BaseInlongGroup {
@Range(min = 0, max = 1, message = "default is 1, only supports [0: disable, 1: enable]")
private Integer enableCreateResource;
- @ApiModelProperty(value = "Standard mode(include Data Ingestion and Synchronization): 0, DataSync mode(only Data Synchronization): 1")
- @Range(min = 0, max = 1, message = "default is 0, only supports [0: Standard, 1: DataSync]")
+ @ApiModelProperty(value = "Standard mode(include Data Ingestion and Synchronization): 0, DataSync mode(only Data Synchronization, real-time data sync in stream way): 1,"
+ + " DataSync mode(only Data Synchronization, offline data sync in batch way): 2")
+ @Range(min = 0, max = 2, message = "default is 0, only supports [0: Standard, 1: DataSync, 2: DataSyncOffline]")
private Integer inlongGroupMode;
@ApiModelProperty(value = "Data report type, default is 0.\n"
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java
index 1b13e941bc..60e5c84835 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/core/impl/AuditServiceImpl.java
@@ -187,7 +187,8 @@ public class AuditServiceImpl implements AuditService {
if (CollectionUtils.isEmpty(request.getAuditIds())) {
// properly overwrite audit ids by role and stream config
- if (InlongConstants.DATASYNC_MODE.equals(groupEntity.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupEntity.getInlongGroupMode())
+ || InlongConstants.DATASYNC_OFFLINE_MODE.equals(groupEntity.getInlongGroupMode())) {
auditIdMap.put(getAuditId(sourceNodeType, IndicatorType.RECEIVED_SUCCESS), sourceNodeType);
request.setAuditIds(getAuditIds(groupId, streamId, sourceNodeType, sinkNodeType));
} else {
@@ -241,7 +242,8 @@ public class AuditServiceImpl implements AuditService {
} else {
auditSet.add(getAuditId(sinkNodeType, IndicatorType.SEND_SUCCESS));
InlongGroupEntity inlongGroup = inlongGroupMapper.selectByGroupId(groupId);
- if (InlongConstants.DATASYNC_MODE.equals(inlongGroup.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(inlongGroup.getInlongGroupMode())
+ || InlongConstants.DATASYNC_OFFLINE_MODE.equals(inlongGroup.getInlongGroupMode())) {
auditSet.add(getAuditId(sourceNodeType, IndicatorType.RECEIVED_SUCCESS));
} else {
auditSet.add(getAuditId(sinkNodeType, IndicatorType.RECEIVED_SUCCESS));
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
index 19e546938f..c9f4019a4d 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/group/InlongGroupServiceImpl.java
@@ -698,7 +698,7 @@ public class InlongGroupServiceImpl implements InlongGroupService {
InlongGroupInfo groupInfo = this.get(groupId);
// check if the group mode is data sync mode
- if (InlongConstants.DATASYNC_MODE.equals(groupInfo.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
String errMSg = String.format("no need to switch sync mode group = {}", groupId);
LOGGER.error(errMSg);
throw new BusinessException(errMSg);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/InitGroupCompleteListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/InitGroupCompleteListener.java
index db8cffbbe1..c96c12c9ff 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/InitGroupCompleteListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/InitGroupCompleteListener.java
@@ -98,7 +98,7 @@ public class InitGroupCompleteListener implements ProcessEventListener {
// update status of other related configs
if (InlongConstants.DISABLE_CREATE_RESOURCE.equals(groupInfo.getEnableCreateResource())) {
- if (InlongConstants.DATASYNC_MODE.equals(groupInfo.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
sourceService.updateStatus(groupId, null, SourceStatus.SOURCE_NORMAL.getCode(), operator);
} else {
sourceService.updateStatus(groupId, null, SourceStatus.TO_BE_ISSUED_ADD.getCode(), operator);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/UpdateGroupCompleteListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/UpdateGroupCompleteListener.java
index 445ee647ef..a6dd8fc786 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/UpdateGroupCompleteListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/group/UpdateGroupCompleteListener.java
@@ -96,7 +96,7 @@ public class UpdateGroupCompleteListener implements ProcessEventListener {
}
// if the inlong group is dataSync mode, the stream source needs to be processed.
- if (InlongConstants.DATASYNC_MODE.equals(groupInfo.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
changeSource4DataSync(groupId, operateType, operator);
}
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java
index 79eb6638cc..f566c29728 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/queue/QueueResourceListener.java
@@ -115,7 +115,7 @@ public class QueueResourceListener implements QueueOperateListener {
String operator = context.getOperator();
GroupOperateType operateType = groupProcessForm.getGroupOperateType();
- if (InlongConstants.DATASYNC_MODE.equals(groupInfo.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
log.warn("skip to execute QueueResourceListener as sync mode for groupId={}", groupId);
if (GroupOperateType.INIT.equals(operateType)) {
this.createQueueForStreams(groupInfo, groupProcessForm.getStreamInfos(), operator);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/InitStreamCompleteListener.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/InitStreamCompleteListener.java
index a739c5da02..df41823b84 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/InitStreamCompleteListener.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/listener/stream/InitStreamCompleteListener.java
@@ -66,7 +66,7 @@ public class InitStreamCompleteListener implements ProcessEventListener {
// Update status of other related configs
streamService.updateStatus(groupId, streamId, StreamStatus.CONFIG_SUCCESSFUL.getCode(), operator);
streamService.updateWithoutCheck(streamInfo.genRequest(), operator);
- if (InlongConstants.DATASYNC_MODE.equals(form.getGroupInfo().getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(form.getGroupInfo().getInlongGroupMode())) {
sourceService.updateStatus(groupId, streamId, SourceStatus.SOURCE_NORMAL.getCode(), operator);
} else {
sourceService.updateStatus(groupId, streamId, SourceStatus.TO_BE_ISSUED_ADD.getCode(), operator);
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
index 9428101c40..83b7ac3ea9 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/AbstractSourceOperator.java
@@ -138,7 +138,7 @@ public abstract class AbstractSourceOperator implements StreamSourceOperator {
updateFieldOpt(entity, request.getFieldList());
return;
}
- boolean allowUpdate = InlongConstants.DATASYNC_MODE.equals(groupMode)
+ boolean allowUpdate = InlongConstants.DATASYNC_REALTIME_MODE.equals(groupMode)
|| SourceStatus.ALLOWED_UPDATE.contains(entity.getStatus());
if (!allowUpdate) {
throw new BusinessException(ErrorCodeEnum.SOURCE_OPT_NOT_ALLOWED,
diff --git a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
index f2a263ba82..1a9dc25243 100644
--- a/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
+++ b/inlong-manager/manager-service/src/main/java/org/apache/inlong/manager/service/source/StreamSourceServiceImpl.java
@@ -190,7 +190,7 @@ public class StreamSourceServiceImpl implements StreamSourceService {
// if the group mode is DATASYNC, just get all related stream sources
List<StreamSource> streamSources = this.listSource(groupId, null);
- if (InlongConstants.DATASYNC_MODE.equals(groupInfo.getInlongGroupMode())) {
+ if (InlongConstants.DATASYNC_REALTIME_MODE.equals(groupInfo.getInlongGroupMode())) {
result = streamSources.stream()
.collect(Collectors.groupingBy(StreamSource::getInlongStreamId, HashMap::new,
Collectors.toCollection(ArrayList::new)));
(inlong) 06/09: [INLONG-10395][Manager] Add interface for schedule client and engine (#10397)
Posted by al...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
aloyszhang pushed a commit to branch dev-offline-sync
in repository https://gitbox.apache.org/repos/asf/inlong.git
commit 1b2e6ef5ca0bbf64a83a2d9f92443a86af6cbee4
Author: AloysZhang <al...@apache.org>
AuthorDate: Thu Jun 13 15:02:14 2024 +0800
[INLONG-10395][Manager] Add interface for schedule client and engine (#10397)
* [INLONG-10395][Manager] Add interface for schedule client and engine
---
inlong-manager/manager-schedule/pom.xml | 41 ++++++++++++++++
.../org/apache/inlong/schedule/ScheduleEngine.java | 54 ++++++++++++++++++++++
.../inlong/schedule/ScheduleEngineClient.java | 45 ++++++++++++++++++
inlong-manager/pom.xml | 1 +
4 files changed, 141 insertions(+)
diff --git a/inlong-manager/manager-schedule/pom.xml b/inlong-manager/manager-schedule/pom.xml
new file mode 100644
index 0000000000..f3a4e0a41f
--- /dev/null
+++ b/inlong-manager/manager-schedule/pom.xml
@@ -0,0 +1,41 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ Licensed to the Apache Software Foundation (ASF) under one or more
+ ~ contributor license agreements. See the NOTICE file distributed with
+ ~ this work for additional information regarding copyright ownership.
+ ~ The ASF licenses this file to You 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
+ ~
+ ~ http://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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>inlong-manager</artifactId>
+ <version>1.13.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>manager-schedule</artifactId>
+
+ <properties>
+ <inlong.root.dir>${project.parent.parent.basedir}</inlong.root.dir>
+ <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
+ </properties>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.inlong</groupId>
+ <artifactId>manager-pojo</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ </dependencies>
+</project>
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngine.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngine.java
new file mode 100644
index 0000000000..bfdb9a88ad
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngine.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+
+/**
+ * Build-in schedule engine, provides basic schedule capabilities.
+ * */
+public interface ScheduleEngine {
+
+ /**
+ * Start schedule engine.
+ * */
+ void start();
+
+ /**
+ * Handle schedule register.
+ * @param scheduleInfo schedule info to register
+ * */
+ boolean handleRegister(ScheduleInfo scheduleInfo);
+
+ /**
+ * Handle schedule unregister.
+ * @param scheduleInfo schedule info to unregister
+ * */
+ boolean handleUnregister(ScheduleInfo scheduleInfo);
+
+ /**
+ * Handle schedule update.
+ * @param scheduleInfo schedule info to update
+ * */
+ boolean handleUpdate(ScheduleInfo scheduleInfo);
+
+ /**
+ * Stop schedule engine.
+ * */
+ void stop();
+}
diff --git a/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngineClient.java b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngineClient.java
new file mode 100644
index 0000000000..1e5ce3460c
--- /dev/null
+++ b/inlong-manager/manager-schedule/src/main/java/org/apache/inlong/schedule/ScheduleEngineClient.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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
+ *
+ * http://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.apache.inlong.schedule;
+
+import org.apache.inlong.manager.pojo.schedule.ScheduleInfo;
+
+/**
+ * Interface for schedule engine client which responses for communicating with schedule engine.
+ * */
+public interface ScheduleEngineClient {
+
+ /**
+ * Register schedule to schedule engine.
+ * @param scheduleInfo schedule info to register
+ * */
+ boolean register(ScheduleInfo scheduleInfo);
+
+ /**
+ * Un-register schedule from schedule engine.
+ * @param scheduleInfo schedule info to unregister
+ * */
+ boolean unregister(ScheduleInfo scheduleInfo);
+
+ /**
+ * Update schedule from schedule engine.
+ * @param scheduleInfo schedule info to update
+ * */
+ boolean update(ScheduleInfo scheduleInfo);
+
+}
diff --git a/inlong-manager/pom.xml b/inlong-manager/pom.xml
index 24288c21c3..bff220728b 100644
--- a/inlong-manager/pom.xml
+++ b/inlong-manager/pom.xml
@@ -41,6 +41,7 @@
<module>manager-workflow</module>
<module>manager-web</module>
<module>manager-docker</module>
+ <module>manager-schedule</module>
</modules>
<properties>