You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by be...@apache.org on 2023/03/19 07:40:17 UTC
[incubator-streampark] branch dev updated: [Feature] Update Variable and set reference Application's release field to NEED_RESTART (#2463)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
The following commit(s) were added to refs/heads/dev by this push:
new 98bc935dd [Feature] Update Variable and set reference Application's release field to NEED_RESTART (#2463)
98bc935dd is described below
commit 98bc935dd01623fe343f11195cfcc159ff0c5eec
Author: gongzhongqiang <76...@qq.com>
AuthorDate: Sun Mar 19 15:40:12 2023 +0800
[Feature] Update Variable and set reference Application's release field to NEED_RESTART (#2463)
* [Feature] Update Variable and set Application's release field to NEED_RESTART
* remove useless code,because list will not be null
---
.../core/controller/VariableController.java | 13 +-----
.../console/core/enums/FlinkAppState.java | 6 +--
.../console/core/service/VariableService.java | 2 +
.../core/service/impl/VariableServiceImpl.java | 54 ++++++++++++++++------
.../src/enums/flinkEnum.ts | 6 +--
.../flink/kubernetes/enums/FlinkJobState.scala | 2 +-
6 files changed, 51 insertions(+), 32 deletions(-)
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/VariableController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/VariableController.java
index 3c068e0c5..9017243ff 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/VariableController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/VariableController.java
@@ -19,7 +19,6 @@ package org.apache.streampark.console.core.controller;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.domain.RestResponse;
-import org.apache.streampark.console.base.exception.ApiAlertException;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.entity.Variable;
import org.apache.streampark.console.core.service.VariableService;
@@ -100,17 +99,7 @@ public class VariableController {
@PutMapping("update")
@RequiresPermissions("variable:update")
public RestResponse updateVariable(@Valid Variable variable) {
- if (variable.getId() == null) {
- throw new ApiAlertException("Sorry, the variable id cannot be null.");
- }
- Variable findVariable = this.variableService.getById(variable.getId());
- if (findVariable == null) {
- throw new ApiAlertException("Sorry, the variable does not exist.");
- }
- if (!findVariable.getVariableCode().equals(variable.getVariableCode())) {
- throw new ApiAlertException("Sorry, the variable code cannot be updated.");
- }
- this.variableService.updateById(variable);
+ variableService.updateVariable(variable);
return RestResponse.success();
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/FlinkAppState.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/FlinkAppState.java
index a46658945..33d53d5db 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/FlinkAppState.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/enums/FlinkAppState.java
@@ -58,7 +58,7 @@ public enum FlinkAppState implements Serializable {
/** Job has been cancelled. */
CANCELED(9),
- /** All of the job's tasks have successfully finished. */
+ /** All the job's tasks have successfully finished. */
FINISHED(10),
/**
@@ -86,10 +86,10 @@ public enum FlinkAppState implements Serializable {
*/
SILENT(17),
- /** Flink job has terminated vaguely, maybe FINISHED, CACNELED or FAILED */
+ /** Flink job has terminated vaguely, maybe FINISHED, CANCELED or FAILED */
TERMINATED(18),
- /** Flink job has terminated vaguely, maybe FINISHED, CACNELED or FAILED */
+ /** Flink job has terminated vaguely, maybe FINISHED, CANCELED or FAILED */
POS_TERMINATED(19),
/** job SUCCEEDED on yarn */
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/VariableService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/VariableService.java
index 073c1bcf9..c255b8254 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/VariableService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/VariableService.java
@@ -70,4 +70,6 @@ public interface VariableService extends IService<Variable> {
String replaceVariable(Long teamId, String mixed);
IPage<Application> dependAppsPage(Variable variable, RestRequest request);
+
+ void updateVariable(Variable variable);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java
index fb58b798d..417d431c2 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/VariableServiceImpl.java
@@ -24,6 +24,7 @@ import org.apache.streampark.console.base.mybatis.pager.MybatisPager;
import org.apache.streampark.console.core.entity.Application;
import org.apache.streampark.console.core.entity.FlinkSql;
import org.apache.streampark.console.core.entity.Variable;
+import org.apache.streampark.console.core.enums.ReleaseState;
import org.apache.streampark.console.core.mapper.VariableMapper;
import org.apache.streampark.console.core.service.ApplicationService;
import org.apache.streampark.console.core.service.CommonService;
@@ -34,6 +35,7 @@ import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.UpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
@@ -110,11 +112,40 @@ public class VariableServiceImpl extends ServiceImpl<VariableMapper, Variable>
page.setTotal(applications.size());
int fromIndex = (request.getPageNum() - 1) * request.getPageSize();
int toIndex = request.getPageNum() * request.getPageSize();
- toIndex = toIndex > applications.size() ? applications.size() : toIndex;
+ toIndex = Math.min(toIndex, applications.size());
page.setRecords(applications.subList(fromIndex, toIndex));
return page;
}
+ @Override
+ public void updateVariable(Variable variable) {
+ // region update variable
+ if (variable.getId() == null) {
+ throw new ApiAlertException("Sorry, the variable id cannot be null.");
+ }
+ Variable findVariable = this.baseMapper.selectById(variable.getId());
+ if (findVariable == null) {
+ throw new ApiAlertException("Sorry, the variable does not exist.");
+ }
+ if (!findVariable.getVariableCode().equals(variable.getVariableCode())) {
+ throw new ApiAlertException("Sorry, the variable code cannot be updated.");
+ }
+ this.baseMapper.updateById(variable);
+ // endregion
+
+ // set Application's field release to NEED_RESTART
+ List<Application> applications = getDependApplicationsByCode(variable);
+ if (CollectionUtils.isNotEmpty(applications)) {
+ applicationService.update(
+ new UpdateWrapper<Application>()
+ .lambda()
+ .in(
+ Application::getId,
+ applications.stream().map(Application::getId).collect(Collectors.toList()))
+ .set(Application::getRelease, ReleaseState.NEED_RESTART.get()));
+ }
+ }
+
@Override
public Variable findByVariableCode(Long teamId, String variableCode) {
LambdaQueryWrapper<Variable> queryWrapper =
@@ -191,22 +222,19 @@ public class VariableServiceImpl extends ServiceImpl<VariableMapper, Variable>
.collect(Collectors.toMap(Application::getId, application -> application));
// Get applications that depend on this variable in application args
- if (applications != null) {
- for (Application app : applications) {
- if (isDepend(variable.getVariableCode(), app.getArgs())) {
- dependApplications.add(app);
- }
+ for (Application app : applications) {
+ if (isDepend(variable.getVariableCode(), app.getArgs())) {
+ dependApplications.add(app);
}
}
+
// Get the application that depends on this variable in flink sql
List<FlinkSql> flinkSqls = flinkSqlService.getByTeamId(variable.getTeamId());
- if (flinkSqls != null) {
- for (FlinkSql flinkSql : flinkSqls) {
- if (isDepend(variable.getVariableCode(), DeflaterUtils.unzipString(flinkSql.getSql()))) {
- Application app = applicationMap.get(flinkSql.getAppId());
- if (!dependApplications.contains(app)) {
- dependApplications.add(applicationMap.get(flinkSql.getAppId()));
- }
+ for (FlinkSql flinkSql : flinkSqls) {
+ if (isDepend(variable.getVariableCode(), DeflaterUtils.unzipString(flinkSql.getSql()))) {
+ Application app = applicationMap.get(flinkSql.getAppId());
+ if (!dependApplications.contains(app)) {
+ dependApplications.add(applicationMap.get(flinkSql.getAppId()));
}
}
}
diff --git a/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts b/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts
index 6a0093b94..2bc19252b 100644
--- a/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts
+++ b/streampark-console/streampark-console-webapp/src/enums/flinkEnum.ts
@@ -96,7 +96,7 @@ export enum AppStateEnum {
CANCELLING = 8,
/** Job has been cancelled. */
CANCELED = 9,
- /** All of the job's tasks have successfully finished. */
+ /** All the job's tasks have successfully finished. */
FINISHED = 10,
/** The job has been suspended which means that it has been stopped but not been removed from a potential HA job store. */
SUSPENDED = 11,
@@ -115,9 +115,9 @@ export enum AppStateEnum {
* A complete loss of flink job tracking translates into LOST state.
*/
SILENT = 17,
- /** Flink job has terminated vaguely, maybe FINISHED, CACNELED or FAILED */
+ /** Flink job has terminated vaguely, maybe FINISHED, CANCELED or FAILED */
TERMINATED = 18,
- /** Flink job has terminated vaguely, maybe FINISHED, CACNELED or FAILED */
+ /** Flink job has terminated vaguely, maybe FINISHED, CANCELED or FAILED */
POS_TERMINATED = 19,
/** job SUCCEEDED on yarn */
SUCCEEDED = 20,
diff --git a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/enums/FlinkJobState.scala b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/enums/FlinkJobState.scala
index 0836bdd2e..353212fc2 100644
--- a/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/enums/FlinkJobState.scala
+++ b/streampark-flink/streampark-flink-kubernetes/src/main/scala/org/apache/streampark/flink/kubernetes/enums/FlinkJobState.scala
@@ -30,7 +30,7 @@ object FlinkJobState extends Enumeration {
SILENT,
// flink job has terminated positively (maybe FINISHED or CANCELED)
POS_TERMINATED,
- // flink job has terminated (maybe FINISHED, CACNELED or FAILED)
+ // flink job has terminated (maybe FINISHED, CANCELED or FAILED)
TERMINATED,
// lost track of flink job completely.
LOST,