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/02/28 00:43:58 UTC
[incubator-streampark] branch dev updated: [Issue-2325] Add ExceptionGetter.java to solve the package order conflict (#2340)
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 831c21cca [Issue-2325] Add ExceptionGetter.java to solve the package order conflict (#2340)
831c21cca is described below
commit 831c21ccac73aa23aa21b9d45345b2f285b651a1
Author: Pan Yuepeng <fl...@126.com>
AuthorDate: Tue Feb 28 08:43:50 2023 +0800
[Issue-2325] Add ExceptionGetter.java to solve the package order conflict (#2340)
[Issue-2325] add methods into Utils.scala for throwing specified exception when check conditions
[Issue-2325] add default exceptoin getter for ApiAlertException.java
[Issue-2325] Change the corresponding lines about exception throw
Remove the redundant added lines
---
.../console/base/exception/ApiAlertException.java | 14 ++
.../interceptor/UploadFileTypeInterceptor.java | 9 +-
.../console/core/controller/ProjectController.java | 5 +-
.../service/impl/ApplicationBackUpServiceImpl.java | 13 +-
.../core/service/impl/ApplicationServiceImpl.java | 220 +++++++++++----------
.../core/service/impl/CommonServiceImpl.java | 7 +-
.../core/service/impl/FlinkEnvServiceImpl.java | 7 +-
.../core/service/impl/ProjectServiceImpl.java | 10 +-
.../core/utils/YarnQueueLabelExpression.java | 23 +--
.../console/system/authentication/JWTFilter.java | 2 +-
.../console/system/authentication/JWTUtil.java | 8 +-
.../console/system/controller/UserController.java | 6 +-
.../system/service/impl/MemberServiceImpl.java | 6 +-
.../system/service/impl/RoleServiceImpl.java | 9 +-
.../system/service/impl/TeamServiceImpl.java | 20 +-
.../system/service/impl/UserServiceImpl.java | 10 +-
16 files changed, 187 insertions(+), 182 deletions(-)
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/exception/ApiAlertException.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/exception/ApiAlertException.java
index a1ec2816c..17ddce28f 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/exception/ApiAlertException.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/exception/ApiAlertException.java
@@ -19,6 +19,8 @@ package org.apache.streampark.console.base.exception;
import org.apache.streampark.console.base.domain.ResponseCode;
+import java.util.Objects;
+
/**
*
*
@@ -43,4 +45,16 @@ public class ApiAlertException extends AbstractApiException {
public ApiAlertException(String message, Throwable cause) {
super(message, cause, ResponseCode.CODE_FAIL_ALERT);
}
+
+ public static void throwIfNull(Object object, String errorMessage) {
+ if (Objects.isNull(object)) {
+ throw new ApiAlertException(errorMessage);
+ }
+ }
+
+ public static void throwIfFalse(boolean expression, String errorMessage) {
+ if (!expression) {
+ throw new ApiAlertException(errorMessage);
+ }
+ }
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/interceptor/UploadFileTypeInterceptor.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/interceptor/UploadFileTypeInterceptor.java
index b205a376e..4af12c99b 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/interceptor/UploadFileTypeInterceptor.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/interceptor/UploadFileTypeInterceptor.java
@@ -18,7 +18,6 @@
package org.apache.streampark.console.base.interceptor;
import org.apache.streampark.common.util.FileUtils;
-import org.apache.streampark.common.util.Utils;
import org.apache.streampark.console.base.exception.ApiAlertException;
import org.springframework.stereotype.Component;
@@ -43,11 +42,11 @@ public class UploadFileTypeInterceptor implements HandlerInterceptor {
Map<String, MultipartFile> files = multipartRequest.getFileMap();
for (String file : files.keySet()) {
MultipartFile multipartFile = multipartRequest.getFile(file);
- Utils.notNull(multipartFile);
+ ApiAlertException.throwIfNull(
+ multipartFile, "File to upload can't be null. Upload file failed.");
boolean fileType = FileUtils.isJarFileType(multipartFile.getInputStream());
- if (!fileType) {
- throw new ApiAlertException("illegal file type, Only standard jar files supported");
- }
+ ApiAlertException.throwIfFalse(
+ fileType, "Illegal file type, Only support standard jar files. Upload file failed.");
}
}
return true;
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProjectController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProjectController.java
index 83ae35b6f..243b9fcfc 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProjectController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/ProjectController.java
@@ -17,9 +17,9 @@
package org.apache.streampark.console.core.controller;
-import org.apache.streampark.common.util.Utils;
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.annotation.AppUpdated;
import org.apache.streampark.console.core.entity.Project;
import org.apache.streampark.console.core.enums.GitAuthorizedError;
@@ -50,7 +50,8 @@ public class ProjectController {
@PostMapping("create")
@RequiresPermissions("project:create")
public RestResponse create(Project project) {
- Utils.notNull(project.getTeamId(), "The teamId cannot be null");
+ ApiAlertException.throwIfNull(
+ project.getTeamId(), "The teamId can't be null. Create team failed.");
return projectService.create(project);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationBackUpServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationBackUpServiceImpl.java
index 2747487a4..d2a48594d 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationBackUpServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationBackUpServiceImpl.java
@@ -19,8 +19,8 @@ package org.apache.streampark.console.core.service.impl;
import org.apache.streampark.common.fs.FsOperator;
import org.apache.streampark.common.util.ThreadUtils;
-import org.apache.streampark.common.util.Utils;
import org.apache.streampark.console.base.domain.RestRequest;
+import org.apache.streampark.console.base.exception.ApiAlertException;
import org.apache.streampark.console.base.exception.InternalException;
import org.apache.streampark.console.base.mybatis.pager.MybatisPager;
import org.apache.streampark.console.core.entity.Application;
@@ -131,12 +131,8 @@ public class ApplicationBackUpServiceImpl
// restore)
fsOperator.delete(application.getAppHome());
- try {
- // copy backup files to a valid dir
- fsOperator.copyDir(backParam.getPath(), application.getAppHome());
- } catch (Exception e) {
- throw e;
- }
+ // copy backup files to a valid dir
+ fsOperator.copyDir(backParam.getPath(), application.getAppHome());
// update restart status
try {
@@ -190,7 +186,8 @@ public class ApplicationBackUpServiceImpl
@Override
public void rollbackFlinkSql(Application application, FlinkSql sql) {
ApplicationBackUp backUp = getFlinkSqlBackup(application.getId(), sql.getId());
- Utils.notNull(backUp);
+ ApiAlertException.throwIfNull(
+ backUp, "Application backup can't be null. Rollback flink sql failed.");
// rollback config and sql
effectiveService.saveOrUpdate(backUp.getAppId(), EffectiveType.CONFIG, backUp.getId());
effectiveService.saveOrUpdate(backUp.getAppId(), EffectiveType.FLINKSQL, backUp.getSqlId());
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
index 6930ecf28..06dbd9aff 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ApplicationServiceImpl.java
@@ -339,9 +339,9 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Override
public void revoke(Application appParma) throws ApplicationException {
Application application = getById(appParma.getId());
- Utils.notNull(
+ ApiAlertException.throwIfNull(
application,
- String.format("The application id=%s cannot be find in the database.", appParma.getId()));
+ String.format("The application id=%s not found, revoke failed.", appParma.getId()));
// 1) delete files that have been published to workspace
application.getFsOperator().delete(application.getAppHome());
@@ -580,8 +580,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Override
public String k8sStartLog(Long id, Integer offset, Integer limit) throws Exception {
Application application = getById(id);
- Utils.notNull(
- application, String.format("The application id=%s cannot be find in the database.", id));
+ ApiAlertException.throwIfNull(
+ application, String.format("The application id=%s can't be found.", id));
if (ExecutionMode.isKubernetesMode(application.getExecutionModeEnum())) {
CompletableFuture<String> future =
CompletableFuture.supplyAsync(
@@ -618,7 +618,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
.get(5, TimeUnit.SECONDS);
} else {
throw new ApiAlertException(
- "job executionMode must be kubernetes-session|kubernetes-application");
+ "Job executionMode must be kubernetes-session|kubernetes-application.");
}
}
@@ -692,7 +692,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Override
@Transactional(rollbackFor = {Exception.class})
public boolean create(Application appParam) {
- Utils.notNull(appParam.getTeamId(), "The teamId cannot be null");
+ ApiAlertException.throwIfNull(
+ appParam.getTeamId(), "The teamId can't be null. Create application failed.");
appParam.setUserId(commonService.getUserId());
appParam.setState(FlinkAppState.ADDED.getValue());
appParam.setRelease(ReleaseState.NEED_RELEASE.get());
@@ -731,9 +732,9 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Transactional(rollbackFor = {Exception.class})
public Long copy(Application appParam) {
boolean existsByJobName = this.existsByJobName(appParam.getJobName());
- if (existsByJobName) {
- throw new ApiAlertException("[StreamPark] Application names cannot be repeated");
- }
+ ApiAlertException.throwIfFalse(
+ !existsByJobName,
+ "[StreamPark] Application names can't be repeated, copy application failed.");
Application oldApp = getById(appParam.getId());
Application newApp = new Application();
@@ -814,110 +815,111 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Override
@Transactional(rollbackFor = {Exception.class})
public boolean update(Application appParam) {
- try {
- checkQueueLabelIfNeed(appParam.getExecutionMode(), appParam.getYarnQueue());
- Application application = getById(appParam.getId());
- application.setRelease(ReleaseState.NEED_RELEASE.get());
- if (application.isUploadJob()) {
- if (!ObjectUtils.safeEquals(application.getJar(), appParam.getJar())) {
- application.setBuild(true);
- } else {
- File jarFile = new File(WebUtils.getAppTempDir(), appParam.getJar());
- if (jarFile.exists()) {
- long checkSum = FileUtils.checksumCRC32(jarFile);
- if (!ObjectUtils.safeEquals(checkSum, application.getJarCheckSum())) {
- application.setBuild(true);
- }
+ checkQueueLabelIfNeed(appParam.getExecutionMode(), appParam.getYarnQueue());
+ Application application = getById(appParam.getId());
+ application.setRelease(ReleaseState.NEED_RELEASE.get());
+ if (application.isUploadJob()) {
+ if (!ObjectUtils.safeEquals(application.getJar(), appParam.getJar())) {
+ application.setBuild(true);
+ } else {
+ File jarFile = new File(WebUtils.getAppTempDir(), appParam.getJar());
+ if (jarFile.exists()) {
+ long checkSum = 0;
+ try {
+ checkSum = FileUtils.checksumCRC32(jarFile);
+ } catch (IOException e) {
+ log.error("Error in checksumCRC32 for {}.", jarFile);
+ throw new RuntimeException(e);
}
- }
- }
-
- if (!application.getBuild()) {
- if (!application.getExecutionMode().equals(appParam.getExecutionMode())) {
- if (appParam.getExecutionModeEnum().equals(ExecutionMode.YARN_APPLICATION)
- || application.getExecutionModeEnum().equals(ExecutionMode.YARN_APPLICATION)) {
+ if (!ObjectUtils.safeEquals(checkSum, application.getJarCheckSum())) {
application.setBuild(true);
}
}
}
+ }
- if (ExecutionMode.isKubernetesMode(appParam.getExecutionMode())) {
- if (!ObjectUtils.safeTrimEquals(
- application.getK8sRestExposedType(), appParam.getK8sRestExposedType())
- || !ObjectUtils.safeTrimEquals(
- application.getK8sJmPodTemplate(), appParam.getK8sJmPodTemplate())
- || !ObjectUtils.safeTrimEquals(
- application.getK8sTmPodTemplate(), appParam.getK8sTmPodTemplate())
- || !ObjectUtils.safeTrimEquals(
- application.getK8sPodTemplates(), appParam.getK8sPodTemplates())
- || !ObjectUtils.safeTrimEquals(
- application.getK8sHadoopIntegration(), appParam.getK8sHadoopIntegration())
- || !ObjectUtils.safeTrimEquals(application.getFlinkImage(), appParam.getFlinkImage())) {
+ if (!application.getBuild()) {
+ if (!application.getExecutionMode().equals(appParam.getExecutionMode())) {
+ if (appParam.getExecutionModeEnum().equals(ExecutionMode.YARN_APPLICATION)
+ || application.getExecutionModeEnum().equals(ExecutionMode.YARN_APPLICATION)) {
application.setBuild(true);
}
}
+ }
- appParam.setJobType(application.getJobType());
- // changes to the following parameters need to be re-release to take effect
- application.setJobName(appParam.getJobName());
- application.setVersionId(appParam.getVersionId());
- application.setArgs(appParam.getArgs());
- application.setOptions(appParam.getOptions());
- application.setDynamicProperties(appParam.getDynamicProperties());
- application.setResolveOrder(appParam.getResolveOrder());
- application.setExecutionMode(appParam.getExecutionMode());
- application.setClusterId(appParam.getClusterId());
- application.setFlinkImage(appParam.getFlinkImage());
- application.setK8sNamespace(appParam.getK8sNamespace());
- application.updateHotParams(appParam);
- application.setK8sRestExposedType(appParam.getK8sRestExposedType());
- application.setK8sPodTemplate(appParam.getK8sPodTemplate());
- application.setK8sJmPodTemplate(appParam.getK8sJmPodTemplate());
- application.setK8sTmPodTemplate(appParam.getK8sTmPodTemplate());
- application.setK8sHadoopIntegration(appParam.getK8sHadoopIntegration());
-
- // changes to the following parameters do not affect running tasks
- application.setModifyTime(new Date());
- application.setDescription(appParam.getDescription());
- application.setAlertId(appParam.getAlertId());
- application.setRestartSize(appParam.getRestartSize());
- application.setCpFailureAction(appParam.getCpFailureAction());
- application.setCpFailureRateInterval(appParam.getCpFailureRateInterval());
- application.setCpMaxFailureInterval(appParam.getCpMaxFailureInterval());
- application.setTags(appParam.getTags());
-
- switch (appParam.getExecutionModeEnum()) {
- case YARN_APPLICATION:
- case YARN_PER_JOB:
- case KUBERNETES_NATIVE_APPLICATION:
- application.setFlinkClusterId(null);
- break;
- case REMOTE:
- case YARN_SESSION:
- case KUBERNETES_NATIVE_SESSION:
- application.setFlinkClusterId(appParam.getFlinkClusterId());
- break;
- default:
- break;
+ if (ExecutionMode.isKubernetesMode(appParam.getExecutionMode())) {
+ if (!ObjectUtils.safeTrimEquals(
+ application.getK8sRestExposedType(), appParam.getK8sRestExposedType())
+ || !ObjectUtils.safeTrimEquals(
+ application.getK8sJmPodTemplate(), appParam.getK8sJmPodTemplate())
+ || !ObjectUtils.safeTrimEquals(
+ application.getK8sTmPodTemplate(), appParam.getK8sTmPodTemplate())
+ || !ObjectUtils.safeTrimEquals(
+ application.getK8sPodTemplates(), appParam.getK8sPodTemplates())
+ || !ObjectUtils.safeTrimEquals(
+ application.getK8sHadoopIntegration(), appParam.getK8sHadoopIntegration())
+ || !ObjectUtils.safeTrimEquals(application.getFlinkImage(), appParam.getFlinkImage())) {
+ application.setBuild(true);
}
+ }
- // Flink Sql job...
- if (application.isFlinkSqlJob()) {
- updateFlinkSqlJob(application, appParam);
+ appParam.setJobType(application.getJobType());
+ // changes to the following parameters need to be re-release to take effect
+ application.setJobName(appParam.getJobName());
+ application.setVersionId(appParam.getVersionId());
+ application.setArgs(appParam.getArgs());
+ application.setOptions(appParam.getOptions());
+ application.setDynamicProperties(appParam.getDynamicProperties());
+ application.setResolveOrder(appParam.getResolveOrder());
+ application.setExecutionMode(appParam.getExecutionMode());
+ application.setClusterId(appParam.getClusterId());
+ application.setFlinkImage(appParam.getFlinkImage());
+ application.setK8sNamespace(appParam.getK8sNamespace());
+ application.updateHotParams(appParam);
+ application.setK8sRestExposedType(appParam.getK8sRestExposedType());
+ application.setK8sPodTemplate(appParam.getK8sPodTemplate());
+ application.setK8sJmPodTemplate(appParam.getK8sJmPodTemplate());
+ application.setK8sTmPodTemplate(appParam.getK8sTmPodTemplate());
+ application.setK8sHadoopIntegration(appParam.getK8sHadoopIntegration());
+
+ // changes to the following parameters do not affect running tasks
+ application.setModifyTime(new Date());
+ application.setDescription(appParam.getDescription());
+ application.setAlertId(appParam.getAlertId());
+ application.setRestartSize(appParam.getRestartSize());
+ application.setCpFailureAction(appParam.getCpFailureAction());
+ application.setCpFailureRateInterval(appParam.getCpFailureRateInterval());
+ application.setCpMaxFailureInterval(appParam.getCpMaxFailureInterval());
+ application.setTags(appParam.getTags());
+
+ switch (appParam.getExecutionModeEnum()) {
+ case YARN_APPLICATION:
+ case YARN_PER_JOB:
+ case KUBERNETES_NATIVE_APPLICATION:
+ application.setFlinkClusterId(null);
+ break;
+ case REMOTE:
+ case YARN_SESSION:
+ case KUBERNETES_NATIVE_SESSION:
+ application.setFlinkClusterId(appParam.getFlinkClusterId());
+ break;
+ default:
+ break;
+ }
+
+ // Flink Sql job...
+ if (application.isFlinkSqlJob()) {
+ updateFlinkSqlJob(application, appParam);
+ } else {
+ if (application.isStreamParkJob()) {
+ configService.update(appParam, application.isRunning());
} else {
- if (application.isStreamParkJob()) {
- configService.update(appParam, application.isRunning());
- } else {
- application.setJar(appParam.getJar());
- application.setMainClass(appParam.getMainClass());
- }
+ application.setJar(appParam.getJar());
+ application.setMainClass(appParam.getMainClass());
}
- baseMapper.updateById(application);
- return true;
- } catch (Exception e) {
- log.error(e.getMessage(), e);
- return false;
}
+ baseMapper.updateById(application);
+ return true;
}
/**
@@ -940,7 +942,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
} else {
// get previous flink sql and decode
FlinkSql copySourceFlinkSql = flinkSqlService.getById(appParam.getSqlId());
- Utils.notNull(copySourceFlinkSql);
+ ApiAlertException.throwIfNull(
+ copySourceFlinkSql, "Flink sql is null, update flink sql job failed.");
copySourceFlinkSql.decode();
// get submit flink sql
@@ -1206,11 +1209,10 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
} else if (ExecutionMode.isYarnMode(application.getExecutionMode())) {
if (ExecutionMode.YARN_SESSION.equals(application.getExecutionModeEnum())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
- Utils.notNull(
+ ApiAlertException.throwIfNull(
cluster,
String.format(
- "The yarn session clusterId=%s cannot be find, maybe the clusterId is wrong or "
- + "the cluster has been deleted. Please contact the Admin.",
+ "The yarn session clusterId=%s can't found, maybe the clusterId is wrong or the cluster has been deleted. Please contact the Admin.",
application.getFlinkClusterId()));
clusterId = cluster.getClusterId();
} else {
@@ -1222,7 +1224,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
if (ExecutionMode.isRemoteMode(application.getExecutionModeEnum())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
- Utils.notNull(
+ ApiAlertException.throwIfNull(
cluster,
String.format(
"The clusterId=%s cannot be find, maybe the clusterId is wrong or "
@@ -1363,7 +1365,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
Application application = getById(appParam.getId());
Utils.notNull(
application,
- String.format("The application id=%s cannot be find in the database.", appParam.getId()));
+ String.format(
+ "The application id=%s not found, start application failed.", appParam.getId()));
application.setState(FlinkAppState.STARTING.getValue());
application.setOptionTime(new Date());
updateById(application);
@@ -1406,7 +1409,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
ApplicationConfig applicationConfig = configService.getEffective(application.getId());
ExecutionMode executionMode = ExecutionMode.of(application.getExecutionMode());
- Utils.notNull(executionMode, "executionMode cannot be null");
+ ApiAlertException.throwIfNull(
+ executionMode, "ExecutionMode can't be null, start application failed.");
if (application.isCustomCodeJob()) {
if (application.isUploadJob()) {
appConf =
@@ -1624,10 +1628,10 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
Map<String, Object> properties = application.getOptionMap();
if (ExecutionMode.isRemoteMode(application.getExecutionModeEnum())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
- Utils.notNull(
+ ApiAlertException.throwIfNull(
cluster,
String.format(
- "The clusterId=%s cannot be find, maybe the clusterId is wrong or "
+ "The clusterId=%s can't be find, maybe the clusterId is wrong or "
+ "the cluster has been deleted. Please contact the Admin.",
application.getFlinkClusterId()));
URI activeAddress = cluster.getRemoteURI();
@@ -1636,7 +1640,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
} else if (ExecutionMode.isYarnMode(application.getExecutionModeEnum())) {
if (ExecutionMode.YARN_SESSION.equals(application.getExecutionModeEnum())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
- Utils.notNull(
+ ApiAlertException.throwIfNull(
cluster,
String.format(
"The yarn session clusterId=%s cannot be find, maybe the clusterId is wrong or "
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/CommonServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/CommonServiceImpl.java
index 05def14df..a4ae29dba 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/CommonServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/CommonServiceImpl.java
@@ -62,9 +62,8 @@ public class CommonServiceImpl implements CommonService {
public String getSqlClientJar(FlinkEnv flinkEnv) {
if (sqlClientJar == null) {
File localClient = WebUtils.getAppClientDir();
- if (!localClient.exists()) {
- throw new ApiAlertException("[StreamPark] " + localClient + " no exists. please check.");
- }
+ ApiAlertException.throwIfFalse(
+ localClient.exists(), "[StreamPark] " + localClient + " no exists. please check.");
List<String> jars =
Arrays.stream(Objects.requireNonNull(localClient.list()))
.filter(
@@ -74,7 +73,7 @@ public class CommonServiceImpl implements CommonService {
.collect(Collectors.toList());
if (jars.isEmpty()) {
throw new ApiAlertException(
- "[StreamPark] can no found streampark-flink-sqlclient jar in " + localClient);
+ "[StreamPark] can't found streampark-flink-sqlclient jar in " + localClient);
}
if (jars.size() > 1) {
throw new ApiAlertException(
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkEnvServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkEnvServiceImpl.java
index 20730d192..f874c1d40 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkEnvServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkEnvServiceImpl.java
@@ -17,6 +17,7 @@
package org.apache.streampark.console.core.service.impl;
+import org.apache.streampark.console.base.exception.ApiAlertException;
import org.apache.streampark.console.core.entity.FlinkEnv;
import org.apache.streampark.console.core.mapper.FlinkEnvMapper;
import org.apache.streampark.console.core.service.FlinkEnvService;
@@ -62,9 +63,9 @@ public class FlinkEnvServiceImpl extends ServiceImpl<FlinkEnvMapper, FlinkEnv>
@Override
public void update(FlinkEnv version) throws IOException {
FlinkEnv flinkEnv = super.getById(version.getId());
- if (flinkEnv == null) {
- throw new RuntimeException("flink home message lost, please check database status!");
- }
+ ApiAlertException.throwIfNull(
+ flinkEnv,
+ "Flink home message lost, update flink env failed, please check database status!");
flinkEnv.setDescription(version.getDescription());
flinkEnv.setFlinkName(version.getFlinkName());
if (!version.getFlinkHome().equals(flinkEnv.getFlinkHome())) {
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
index 7c1091434..fa80e49c4 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/ProjectServiceImpl.java
@@ -117,8 +117,9 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
try {
Project project = getById(projectParam.getId());
Utils.notNull(project);
- Utils.required(
- project.getTeamId().equals(projectParam.getTeamId()), "TeamId cannot be changed.");
+ ApiAlertException.throwIfFalse(
+ project.getTeamId().equals(projectParam.getTeamId()),
+ "TeamId can't be changed, update project failed.");
project.setName(projectParam.getName());
project.setUrl(projectParam.getUrl());
project.setBranches(projectParam.getBranches());
@@ -258,9 +259,8 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
@Override
public List<String> jars(Project project) {
List<String> list = new ArrayList<>(0);
- if (project.getModule() == null) {
- throw new ApiAlertException("project module can not be null, please check");
- }
+ ApiAlertException.throwIfNull(
+ project.getModule(), "Project module can't be null, please check.");
File apps = new File(project.getDistHome(), project.getModule());
for (File file : Objects.requireNonNull(apps.listFiles())) {
if (file.getName().endsWith(".jar")) {
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpression.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpression.java
index 370005cb4..a3e77d885 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpression.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/utils/YarnQueueLabelExpression.java
@@ -19,6 +19,7 @@ package org.apache.streampark.console.core.utils;
import org.apache.streampark.common.conf.ConfigConst;
import org.apache.streampark.common.enums.ExecutionMode;
+import org.apache.streampark.console.base.exception.ApiAlertException;
import com.google.common.annotations.VisibleForTesting;
import org.apache.commons.lang3.StringUtils;
@@ -90,32 +91,26 @@ public class YarnQueueLabelExpression {
*/
public static void checkQueueLabelIfNeed(int executionMode, String queueLabel) {
if (ExecutionMode.isYarnMode(executionMode)) {
- if (!YarnQueueLabelExpression.isValid(queueLabel, true)) {
- throw new IllegalArgumentException(ERR_HINTS);
- }
+ ApiAlertException.throwIfFalse(isValid(queueLabel, true), ERR_HINTS);
}
}
// Visible for test.
public static YarnQueueLabelExpression of(@Nonnull String queueLabelExpr) {
- if (isValid(queueLabelExpr, false)) {
- String[] strs = queueLabelExpr.split(AT);
- if (strs.length == 2) {
- return new YarnQueueLabelExpression(strs[0], strs[1]);
- }
- return new YarnQueueLabelExpression(strs[0], null);
+ ApiAlertException.throwIfFalse(isValid(queueLabelExpr, false), ERR_HINTS);
+ String[] strs = queueLabelExpr.split(AT);
+ if (strs.length == 2) {
+ return new YarnQueueLabelExpression(strs[0], strs[1]);
}
- throw new IllegalArgumentException(ERR_HINTS);
+ return new YarnQueueLabelExpression(strs[0], null);
}
public static YarnQueueLabelExpression of(
@Nonnull String queue, @Nullable String labelExpression) {
YarnQueueLabelExpression queueLabelExpression =
new YarnQueueLabelExpression(queue, labelExpression);
- if (isValid(queueLabelExpression.toString(), false)) {
- return queueLabelExpression;
- }
- throw new IllegalArgumentException(ERR_HINTS);
+ ApiAlertException.throwIfFalse(isValid(queueLabelExpression.toString(), false), ERR_HINTS);
+ return queueLabelExpression;
}
public static Map<String, String> getQueueLabelMap(String queueLabelExp) {
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/JWTFilter.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/JWTFilter.java
index dfb559215..103dc04af 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/JWTFilter.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/JWTFilter.java
@@ -80,7 +80,7 @@ public class JWTFilter extends BasicHttpAuthenticationFilter {
getSubject(request, response).login(jwtToken);
return true;
} catch (Exception e) {
- log.info(e.getMessage());
+ log.error("Error in executeLogin, token {}, jwtToken {}", token, jwtToken, e);
return false;
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/JWTUtil.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/JWTUtil.java
index c4105bf22..dc5d207de 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/JWTUtil.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/authentication/JWTUtil.java
@@ -53,7 +53,7 @@ public class JWTUtil {
} catch (TokenExpiredException e) {
throw new AuthenticationException(e.getMessage());
} catch (Exception e) {
- log.info("token is invalid:{} , e:{}", e.getMessage(), e.getClass());
+ log.error("token is invalid:{} , e:{}", e.getMessage(), e.getClass());
return false;
}
}
@@ -64,7 +64,7 @@ public class JWTUtil {
DecodedJWT jwt = JWT.decode(token);
return jwt.getClaim("userName").asString();
} catch (JWTDecodeException e) {
- log.info("error：{}", e.getMessage());
+ log.error("error：{}", e.getMessage());
return null;
}
}
@@ -74,7 +74,7 @@ public class JWTUtil {
DecodedJWT jwt = JWT.decode(token);
return jwt.getClaim("userId").asLong();
} catch (JWTDecodeException e) {
- log.info("error：{}", e.getMessage());
+ log.error("error：{}", e.getMessage());
return null;
}
}
@@ -110,7 +110,7 @@ public class JWTUtil {
.withExpiresAt(date)
.sign(algorithm);
} catch (Exception e) {
- log.info("error：{}", e);
+ log.error("error：{}", e);
return null;
}
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/UserController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/UserController.java
index 09b5cf1d0..238fe6f5f 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/UserController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/controller/UserController.java
@@ -171,12 +171,10 @@ public class UserController {
public RestResponse setTeam(Long teamId) {
Team team = teamService.getById(teamId);
if (team == null) {
- return RestResponse.fail("teamId is invalid", ResponseCode.CODE_FAIL_ALERT);
+ return RestResponse.fail("TeamId is invalid, set team failed.", ResponseCode.CODE_FAIL_ALERT);
}
User user = commonService.getCurrentUser();
- if (user == null) {
- throw new ApiAlertException("current login user is null, set team failed");
- }
+ ApiAlertException.throwIfNull(user, "Current login user is null, set team failed.");
// 1) set the latest team
userService.setLastTeam(teamId, user.getUserId());
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/MemberServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/MemberServiceImpl.java
index 420911233..56a3b3e18 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/MemberServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/MemberServiceImpl.java
@@ -75,7 +75,7 @@ public class MemberServiceImpl extends ServiceImpl<MemberMapper, Member> impleme
@Override
public IPage<Member> findUsers(Member member, RestRequest request) {
- Utils.notNull(member.getTeamId(), "The team id is required.");
+ ApiAlertException.throwIfNull(member.getTeamId(), "The team id is required.");
Page<Member> page = new Page<>();
page.setCurrent(request.getPageNum());
page.setSize(request.getPageSize());
@@ -102,7 +102,7 @@ public class MemberServiceImpl extends ServiceImpl<MemberMapper, Member> impleme
}
private Member findByUserId(Long teamId, Long userId) {
- Utils.notNull(teamId, "The team id is required.");
+ ApiAlertException.throwIfNull(teamId, "The team id is required.");
LambdaQueryWrapper<Member> queryWrapper =
new LambdaQueryWrapper<Member>()
.eq(Member::getTeamId, teamId)
@@ -137,7 +137,7 @@ public class MemberServiceImpl extends ServiceImpl<MemberMapper, Member> impleme
() ->
new ApiAlertException(
String.format("The teamId [%s] not found", member.getTeamId())));
- Utils.required(
+ ApiAlertException.throwIfFalse(
findByUserId(member.getTeamId(), user.getUserId()) == null,
String.format(
"The user [%s] has been added the team [%s], please don't add it again.",
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/RoleServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/RoleServiceImpl.java
index 1ca9dbdb2..0212dd166 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/RoleServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/RoleServiceImpl.java
@@ -17,9 +17,9 @@
package org.apache.streampark.console.system.service.impl;
-import org.apache.streampark.common.util.Utils;
import org.apache.streampark.console.base.domain.Constant;
import org.apache.streampark.console.base.domain.RestRequest;
+import org.apache.streampark.console.base.exception.ApiAlertException;
import org.apache.streampark.console.system.entity.Role;
import org.apache.streampark.console.system.entity.RoleMenu;
import org.apache.streampark.console.system.mapper.RoleMapper;
@@ -84,12 +84,13 @@ public class RoleServiceImpl extends ServiceImpl<RoleMapper, Role> implements Ro
Optional.ofNullable(this.getById(roleId))
.orElseThrow(
() ->
- new IllegalArgumentException(String.format("Role id [%s] not found", roleId)));
+ new ApiAlertException(
+ String.format("Role id [%s] not found. Delete role failed.", roleId)));
List<Long> userIdsByRoleId = memberService.findUserIdsByRoleId(roleId);
- Utils.required(
+ ApiAlertException.throwIfFalse(
userIdsByRoleId == null || userIdsByRoleId.isEmpty(),
String.format(
- "There are some users are bound to role %s , please unbind it first.",
+ "There are some users of role %s, delete role failed, please unbind it first.",
role.getRoleName()));
this.removeById(roleId);
this.roleMenuService.deleteByRoleId(roleId);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/TeamServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/TeamServiceImpl.java
index a786de301..e4899831c 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/TeamServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/TeamServiceImpl.java
@@ -17,7 +17,6 @@
package org.apache.streampark.console.system.service.impl;
-import org.apache.streampark.common.util.Utils;
import org.apache.streampark.console.base.domain.RestRequest;
import org.apache.streampark.console.base.exception.ApiAlertException;
import org.apache.streampark.console.core.enums.UserType;
@@ -81,11 +80,11 @@ public class TeamServiceImpl extends ServiceImpl<TeamMapper, Team> implements Te
@Override
public void createTeam(Team team) {
Team existedTeam = findByName(team.getTeamName());
- if (existedTeam != null) {
- throw new IllegalArgumentException(
- String.format(
- "Team name [%s] was found, please rename and try again.", team.getTeamName()));
- }
+ ApiAlertException.throwIfFalse(
+ existedTeam == null,
+ String.format(
+ "Team name [%s] exists already. Create team failed. Please rename and try again.",
+ team.getTeamName()));
team.setId(null);
team.setCreateTime(new Date());
team.setModifyTime(team.getCreateTime());
@@ -129,8 +128,9 @@ public class TeamServiceImpl extends ServiceImpl<TeamMapper, Team> implements Te
() ->
new IllegalArgumentException(
String.format("Team id [id=%s] not found", team.getId())));
- Utils.required(
- oldTeam.getTeamName().equals(team.getTeamName()), "Team name cannot be changed.");
+ ApiAlertException.throwIfFalse(
+ oldTeam.getTeamName().equals(team.getTeamName()),
+ "Team name can't be changed. Update team failed.");
oldTeam.setDescription(team.getDescription());
oldTeam.setModifyTime(new Date());
updateById(oldTeam);
@@ -141,9 +141,7 @@ public class TeamServiceImpl extends ServiceImpl<TeamMapper, Team> implements Te
User user =
Optional.ofNullable(userService.getById(userId))
.orElseThrow(
- () ->
- new IllegalArgumentException(
- String.format("The userId [%s] not found", userId)));
+ () -> new ApiAlertException(String.format("The userId [%s] not found.", userId)));
// Admin has the permission for all teams.
if (UserType.ADMIN.equals(user.getUserType())) {
return this.list();
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/UserServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/UserServiceImpl.java
index e60472edc..354a239f3 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/UserServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/system/service/impl/UserServiceImpl.java
@@ -138,14 +138,12 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
@Transactional(rollbackFor = Exception.class)
public void updatePassword(User userParam) {
User user = getById(userParam.getUserId());
- if (user == null) {
- throw new ApiAlertException("update password failed, user is null");
- }
+ ApiAlertException.throwIfNull(user, "User is null. Update password failed.");
String saltPassword = ShaHashUtils.encrypt(user.getSalt(), userParam.getOldPassword());
- if (!StringUtils.equals(user.getPassword(), saltPassword)) {
- throw new ApiAlertException("update password failed, old password error");
- }
+ ApiAlertException.throwIfFalse(
+ StringUtils.equals(user.getPassword(), saltPassword),
+ "Old password error. Update password failed.");
String salt = ShaHashUtils.getRandomSalt();
String password = ShaHashUtils.encrypt(salt, userParam.getPassword());