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/15 17:50:30 UTC
[incubator-streampark] branch utils created (now de9759d7b)
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a change to branch utils
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
at de9759d7b [Improve] call utils.required improvement
This branch includes the following new commits:
new de9759d7b [Improve] call utils.required improvement
The 1 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.
[incubator-streampark] 01/01: [Improve] call utils.required improvement
Posted by be...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
benjobs pushed a commit to branch utils
in repository https://gitbox.apache.org/repos/asf/incubator-streampark.git
commit de9759d7b783d0cabdd7a459ba7b3c4955e5fed9
Author: benjobs <be...@apache.org>
AuthorDate: Thu Feb 16 00:55:01 2023 +0800
[Improve] call utils.required improvement
---
.../org/apache/streampark/common/util/Utils.scala | 2 +-
.../interceptor/UploadFileTypeInterceptor.java | 2 +-
.../streampark/console/base/util/CommonUtils.java | 2 +-
.../console/core/controller/ProjectController.java | 2 +-
.../console/core/runner/EnvInitializer.java | 2 +-
.../core/service/alert/impl/AlertServiceImpl.java | 2 +-
.../core/service/impl/AppBuildPipeServiceImpl.java | 2 +-
.../service/impl/ApplicationBackUpServiceImpl.java | 2 +-
.../core/service/impl/ApplicationServiceImpl.java | 59 +++++++++++-----------
.../core/service/impl/FlinkSqlServiceImpl.java | 4 +-
.../core/service/impl/ProjectServiceImpl.java | 8 +--
.../core/service/impl/SavePointServiceImpl.java | 2 +-
.../core/service/impl/SqlCompleteServiceImpl.java | 2 +-
.../console/core/task/FlinkRESTAPIWatcher.java | 2 -
.../system/service/impl/MemberServiceImpl.java | 4 +-
.../system/service/impl/UserServiceImpl.java | 6 +--
.../connector/hbase/source/HBaseJavaSource.java | 4 +-
.../flink/connector/jdbc/sink/JdbcJavaSink.java | 2 +-
.../connector/jdbc/source/JdbcJavaSource.java | 4 +-
.../connector/mongo/source/MongoJavaSource.java | 6 +--
20 files changed, 58 insertions(+), 61 deletions(-)
diff --git a/streampark-common/src/main/scala/org/apache/streampark/common/util/Utils.scala b/streampark-common/src/main/scala/org/apache/streampark/common/util/Utils.scala
index a83314d93..1737c48ec 100644
--- a/streampark-common/src/main/scala/org/apache/streampark/common/util/Utils.scala
+++ b/streampark-common/src/main/scala/org/apache/streampark/common/util/Utils.scala
@@ -35,7 +35,7 @@ object Utils {
}
def notNull(obj: Any): Unit = {
- notNull(obj, "this argument is required; it must not be null")
+ notNull(obj, "this argument must not be null")
}
def notEmpty(elem: Any): Boolean = {
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 69c96e0b3..b205a376e 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
@@ -43,7 +43,7 @@ public class UploadFileTypeInterceptor implements HandlerInterceptor {
Map<String, MultipartFile> files = multipartRequest.getFileMap();
for (String file : files.keySet()) {
MultipartFile multipartFile = multipartRequest.getFile(file);
- Utils.required(multipartFile != null);
+ Utils.notNull(multipartFile);
boolean fileType = FileUtils.isJarFileType(multipartFile.getInputStream());
if (!fileType) {
throw new ApiAlertException("illegal file type, Only standard jar files supported");
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/CommonUtils.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/CommonUtils.java
index 0efb46a12..477443b6f 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/CommonUtils.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/base/util/CommonUtils.java
@@ -451,7 +451,7 @@ public final class CommonUtils implements Serializable {
}
public static <T> T[] arrayRemoveElements(T[] array, T... elem) {
- Utils.required(array != null);
+ Utils.notNull(array);
List<T> arrayList = new ArrayList<>(0);
Collections.addAll(arrayList, array);
if (isEmpty(elem)) {
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 0f5818dbc..83ae35b6f 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
@@ -50,7 +50,7 @@ public class ProjectController {
@PostMapping("create")
@RequiresPermissions("project:create")
public RestResponse create(Project project) {
- Utils.required(project.getTeamId() != null, "The teamId cannot be null");
+ Utils.notNull(project.getTeamId(), "The teamId cannot be null");
return projectService.create(project);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
index 9327d6f70..69436a320 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/runner/EnvInitializer.java
@@ -98,7 +98,7 @@ public class EnvInitializer implements ApplicationRunner {
.forEach(
key -> {
InternalOption config = InternalConfigHolder.getConfig(key);
- Utils.required(config != null);
+ Utils.notNull(config);
InternalConfigHolder.set(config, springEnv.getProperty(key, config.classType()));
});
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/AlertServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/AlertServiceImpl.java
index 7d7beb456..eb7600ef7 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/AlertServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/alert/impl/AlertServiceImpl.java
@@ -84,7 +84,7 @@ public class AlertServiceImpl implements AlertService {
try {
Class<? extends AlertNotifyService> notifyServiceClass =
getAlertServiceImpl(alertType);
- Utils.required(notifyServiceClass != null);
+ Utils.notNull(notifyServiceClass);
boolean alertRes =
SpringContextUtils.getBean(notifyServiceClass)
.doAlert(params, alertTemplate);
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
index 9d4134d72..007eba681 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/AppBuildPipeServiceImpl.java
@@ -145,7 +145,7 @@ public class AppBuildPipeServiceImpl
FlinkSql effectiveFlinkSql = flinkSqlService.getEffective(app.getId(), false);
if (app.isFlinkSqlJob()) {
FlinkSql flinkSql = newFlinkSql == null ? effectiveFlinkSql : newFlinkSql;
- Utils.required(flinkSql != null);
+ Utils.notNull(flinkSql);
app.setDependency(flinkSql.getDependency());
}
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 8902c3f64..325c5735b 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
@@ -190,7 +190,7 @@ public class ApplicationBackUpServiceImpl
@Override
public void rollbackFlinkSql(Application application, FlinkSql sql) {
ApplicationBackUp backUp = getFlinkSqlBackup(application.getId(), sql.getId());
- Utils.required(backUp != null);
+ Utils.notNull(backUp);
// 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 d51002325..614acc388 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
@@ -337,8 +337,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Override
public void revoke(Application appParma) throws ApplicationException {
Application application = getById(appParma.getId());
- Utils.required(
- application != null,
+ Utils.notNull(
+ application,
String.format("The application id=%s cannot be find in the database.", appParma.getId()));
// 1) delete files that have been published to workspace
@@ -578,9 +578,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.required(
- application != null,
- String.format("The application id=%s cannot be find in the database.", id));
+ Utils.notNull(
+ application, String.format("The application id=%s cannot be find in the database.", id));
if (ExecutionMode.isKubernetesMode(application.getExecutionModeEnum())) {
CompletableFuture<String> future =
CompletableFuture.supplyAsync(
@@ -691,7 +690,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Override
@Transactional(rollbackFor = {Exception.class})
public boolean create(Application appParam) {
- Utils.required(appParam.getTeamId() != null, "The teamId cannot be null");
+ Utils.notNull(appParam.getTeamId(), "The teamId cannot be null");
appParam.setUserId(commonService.getUserId());
appParam.setState(FlinkAppState.ADDED.getValue());
appParam.setLaunch(LaunchState.NEED_LAUNCH.get());
@@ -705,8 +704,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
appParam.setJarCheckSum(FileUtils.checksumCRC32(new File(jarPath)));
}
- boolean saved = save(appParam);
- if (saved) {
+ if (save(appParam)) {
if (appParam.isFlinkSqlJob()) {
FlinkSql flinkSql = new FlinkSql(appParam);
flinkSqlService.create(flinkSql);
@@ -714,10 +712,10 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
if (appParam.getConfig() != null) {
configService.create(appParam, true);
}
- Utils.required(appParam.getId() != null);
return true;
+ } else {
+ throw new ApiAlertException("create application failed");
}
- return false;
}
private boolean existsByJobName(String jobName) {
@@ -803,10 +801,11 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
configService.save(config);
configService.setLatestOrEffective(true, config.getId(), newApp.getId());
}
- Utils.required(newApp.getId() != null);
return newApp.getId();
+ } else {
+ throw new ApiAlertException(
+ "create application from copy failed, copy source app: " + oldApp.getJobName());
}
- return 0L;
}
@Override
@@ -937,7 +936,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
} else {
// get previous flink sql and decode
FlinkSql copySourceFlinkSql = flinkSqlService.getById(appParam.getSqlId());
- Utils.required(copySourceFlinkSql != null);
+ Utils.notNull(copySourceFlinkSql);
copySourceFlinkSql.decode();
// get submit flink sql
@@ -1192,8 +1191,8 @@ 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.required(
- cluster != null,
+ Utils.notNull(
+ 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.",
@@ -1208,8 +1207,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
if (ExecutionMode.isRemoteMode(application.getExecutionModeEnum())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
- Utils.required(
- cluster != null,
+ Utils.notNull(
+ cluster,
String.format(
"The clusterId=%s cannot be find, maybe the clusterId is wrong or "
+ "the cluster has been deleted. Please contact the Admin.",
@@ -1347,8 +1346,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
@Override
public void starting(Application appParam) {
Application application = getById(appParam.getId());
- Utils.required(
- application != null,
+ Utils.notNull(
+ application,
String.format("The application id=%s cannot be find in the database.", appParam.getId()));
application.setState(FlinkAppState.STARTING.getValue());
application.setOptionTime(new Date());
@@ -1361,7 +1360,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
final Application application = getById(appParam.getId());
- Utils.required(application != null);
+ Utils.notNull(application);
application.setAllowNonRestored(appParam.getAllowNonRestored());
FlinkEnv flinkEnv = flinkEnvService.getByIdOrDefault(application.getVersionId());
@@ -1392,7 +1391,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
ApplicationConfig applicationConfig = configService.getEffective(application.getId());
ExecutionMode executionMode = ExecutionMode.of(application.getExecutionMode());
- Utils.required(executionMode != null, "executionMode cannot be null");
+ Utils.notNull(executionMode, "executionMode cannot be null");
if (application.isCustomCodeJob()) {
if (application.isUploadJob()) {
appConf =
@@ -1440,7 +1439,7 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
}
} else if (application.isFlinkSqlJob()) {
FlinkSql flinkSql = flinkSqlService.getEffective(application.getId(), false);
- Utils.required(flinkSql != null);
+ Utils.notNull(flinkSql);
// 1) dist_userJar
String sqlDistJar = commonService.getSqlClientJar(flinkEnv);
// 2) appConfig
@@ -1474,14 +1473,14 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
AppBuildPipeline buildPipeline = appBuildPipeService.getById(application.getId());
- Utils.required(buildPipeline != null);
+ Utils.notNull(buildPipeline);
BuildResult buildResult = buildPipeline.getBuildResult();
if (ExecutionMode.YARN_APPLICATION.equals(executionMode)) {
buildResult = new ShadedBuildResponse(null, flinkUserJar, true);
} else {
if (ExecutionMode.isKubernetesApplicationMode(application.getExecutionMode())) {
- Utils.required(buildResult != null);
+ Utils.notNull(buildResult);
DockerImageBuildResponse result = buildResult.as(DockerImageBuildResponse.class);
String ingressTemplates = application.getIngressTemplate();
String domainName = application.getDefaultModeIngress();
@@ -1610,8 +1609,8 @@ 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.required(
- cluster != null,
+ Utils.notNull(
+ cluster,
String.format(
"The clusterId=%s cannot be find, maybe the clusterId is wrong or "
+ "the cluster has been deleted. Please contact the Admin.",
@@ -1627,8 +1626,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
}
if (ExecutionMode.YARN_SESSION.equals(application.getExecutionModeEnum())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
- Utils.required(
- cluster != null,
+ Utils.notNull(
+ 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.",
@@ -1719,8 +1718,8 @@ public class ApplicationServiceImpl extends ServiceImpl<ApplicationMapper, Appli
// 3.1) At the remote mode, request the flink webui interface to get the savepoint path
if (ExecutionMode.isRemoteMode(application.getExecutionMode())) {
FlinkCluster cluster = flinkClusterService.getById(application.getFlinkClusterId());
- Utils.required(
- cluster != null,
+ Utils.notNull(
+ cluster,
String.format(
"The clusterId=%s cannot be find, maybe the clusterId is wrong or "
+ "the cluster has been deleted. Please contact the Admin.",
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSqlServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSqlServiceImpl.java
index ba0b3bd05..7e662a75b 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSqlServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/FlinkSqlServiceImpl.java
@@ -169,11 +169,11 @@ public class FlinkSqlServiceImpl extends ServiceImpl<FlinkSqlMapper, FlinkSql>
@Transactional(propagation = Propagation.REQUIRES_NEW, rollbackFor = Exception.class)
public void rollback(Application application) {
FlinkSql sql = getCandidate(application.getId(), CandidateType.HISTORY);
- Utils.required(sql != null);
+ Utils.notNull(sql);
try {
// check and backup current job
FlinkSql effectiveSql = getEffective(application.getId(), false);
- Utils.required(effectiveSql != null);
+ Utils.notNull(effectiveSql);
// rollback history sql
backUpService.rollbackFlinkSql(application, sql);
} catch (Exception e) {
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 fd9112e0e..4b87409c9 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
@@ -116,7 +116,7 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
public boolean update(Project projectParam) {
try {
Project project = getById(projectParam.getId());
- Utils.required(project != null);
+ Utils.notNull(project);
Utils.required(
project.getTeamId().equals(projectParam.getTeamId()), "TeamId cannot be changed.");
project.setName(projectParam.getName());
@@ -162,7 +162,7 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
@Transactional(rollbackFor = {Exception.class})
public boolean delete(Long id) {
Project project = getById(id);
- Utils.required(project != null);
+ Utils.notNull(project);
LambdaQueryWrapper<Application> queryWrapper =
new LambdaQueryWrapper<Application>().eq(Application::getProjectId, id);
long count = applicationService.count(queryWrapper);
@@ -234,7 +234,7 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
@Override
public List<String> modules(Long id) {
Project project = getById(id);
- Utils.required(project != null);
+ Utils.notNull(project);
BuildState buildState = BuildState.of(project.getBuildState());
if (BuildState.SUCCESSFUL.equals(buildState)) {
File appHome = project.getDistHome();
@@ -309,7 +309,7 @@ public class ProjectServiceImpl extends ServiceImpl<ProjectMapper, Project>
}
List<Map<String, Object>> list = new ArrayList<>();
File[] files = unzipFile.listFiles(x -> "conf".equals(x.getName()));
- Utils.required(files != null);
+ Utils.notNull(files);
for (File item : files) {
eachFile(item, list, true);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java
index 2aebae099..78261a699 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SavePointServiceImpl.java
@@ -67,7 +67,7 @@ public class SavePointServiceImpl extends ServiceImpl<SavePointMapper, SavePoint
private void expire(SavePoint entity) {
FlinkEnv flinkEnv = flinkEnvService.getByAppId(entity.getAppId());
- Utils.required(flinkEnv != null);
+ Utils.notNull(flinkEnv);
int cpThreshold =
Integer.parseInt(
flinkEnv.convertFlinkYamlAsMap().getOrDefault("state.checkpoints.num-retained", "5"));
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SqlCompleteServiceImpl.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SqlCompleteServiceImpl.java
index ddeeb22e4..5ce506db8 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SqlCompleteServiceImpl.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/impl/SqlCompleteServiceImpl.java
@@ -185,7 +185,7 @@ public class SqlCompleteServiceImpl implements SqlCompleteService {
nowStep = nowStep.get(nowChar).getNext();
loc += 1;
}
- Utils.required(preNode != null);
+ Utils.notNull(preNode);
preNode.setStop();
preNode.setCount(count);
}
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkRESTAPIWatcher.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkRESTAPIWatcher.java
index 9edab9681..6d82c2316 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkRESTAPIWatcher.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/task/FlinkRESTAPIWatcher.java
@@ -20,7 +20,6 @@ package org.apache.streampark.console.core.task;
import org.apache.streampark.common.enums.ExecutionMode;
import org.apache.streampark.common.util.HttpClientUtils;
import org.apache.streampark.common.util.ThreadUtils;
-import org.apache.streampark.common.util.Utils;
import org.apache.streampark.common.util.YarnUtils;
import org.apache.streampark.console.base.util.JacksonUtils;
import org.apache.streampark.console.core.entity.Application;
@@ -207,7 +206,6 @@ public class FlinkRESTAPIWatcher {
final OptionState optionState = OPTIONING.get(key);
try {
// query status from flink rest api
- Utils.required(application.getId() != null);
getFromFlinkRestApi(application, stopFrom);
} catch (Exception flinkException) {
// query status from yarn rest api
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 c4f015e3e..420911233 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.required(member.getTeamId() != null, "The team id is required.");
+ Utils.notNull(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.required(teamId != null, "The team id is required.");
+ Utils.notNull(teamId, "The team id is required.");
LambdaQueryWrapper<Member> queryWrapper =
new LambdaQueryWrapper<Member>()
.eq(Member::getTeamId, teamId)
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 99ba05aad..e60472edc 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
@@ -74,7 +74,7 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
page.setSize(request.getPageSize());
IPage<User> resPage = this.baseMapper.findUserDetail(page, user);
- Utils.required(resPage != null);
+ Utils.notNull(resPage);
if (resPage.getTotal() == 0) {
resPage.setRecords(Collections.emptyList());
}
@@ -189,7 +189,7 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
@Override
public void setLastTeam(Long teamId, Long userId) {
User user = getById(userId);
- Utils.required(user != null);
+ Utils.notNull(user);
user.setLastTeamId(teamId);
this.baseMapper.updateById(user);
}
@@ -197,7 +197,7 @@ public class UserServiceImpl extends ServiceImpl<UserMapper, User> implements Us
@Override
public void clearLastTeam(Long userId, Long teamId) {
User user = getById(userId);
- Utils.required(user != null);
+ Utils.notNull(user);
if (!teamId.equals(user.getLastTeamId())) {
return;
}
diff --git a/streampark-flink/streampark-flink-connector/streampark-flink-connector-hbase/src/main/scala/org/apache/streampark/flink/connector/hbase/source/HBaseJavaSource.java b/streampark-flink/streampark-flink-connector/streampark-flink-connector-hbase/src/main/scala/org/apache/streampark/flink/connector/hbase/source/HBaseJavaSource.java
index 4d8b59314..c517f72e3 100644
--- a/streampark-flink/streampark-flink-connector/streampark-flink-connector-hbase/src/main/scala/org/apache/streampark/flink/connector/hbase/source/HBaseJavaSource.java
+++ b/streampark-flink/streampark-flink-connector/streampark-flink-connector-hbase/src/main/scala/org/apache/streampark/flink/connector/hbase/source/HBaseJavaSource.java
@@ -41,8 +41,8 @@ public class HBaseJavaSource<T> {
HBaseResultFunction<T> resultFunction,
RunningFunction runningFunc) {
- Utils.required(queryFunction != null, "queryFunction must not be null");
- Utils.required(resultFunction != null, "resultFunction must not be null");
+ Utils.notNull(queryFunction, "queryFunction must not be null");
+ Utils.notNull(resultFunction, "resultFunction must not be null");
HBaseSourceFunction<T> sourceFunction = new HBaseSourceFunction<>(
property,
queryFunction,
diff --git a/streampark-flink/streampark-flink-connector/streampark-flink-connector-jdbc/src/main/scala/org/apache/streampark/flink/connector/jdbc/sink/JdbcJavaSink.java b/streampark-flink/streampark-flink-connector/streampark-flink-connector-jdbc/src/main/scala/org/apache/streampark/flink/connector/jdbc/sink/JdbcJavaSink.java
index 1bbb0cc8a..34296d55c 100644
--- a/streampark-flink/streampark-flink-connector/streampark-flink-connector-jdbc/src/main/scala/org/apache/streampark/flink/connector/jdbc/sink/JdbcJavaSink.java
+++ b/streampark-flink/streampark-flink-connector/streampark-flink-connector-jdbc/src/main/scala/org/apache/streampark/flink/connector/jdbc/sink/JdbcJavaSink.java
@@ -55,7 +55,7 @@ public class JdbcJavaSink<T> {
}
public DataStreamSink<T> sink(DataStream<T> dataStream) {
- Utils.required(sqlFunc != null, "transformFunction can not be null");
+ Utils.notNull(sqlFunc, "transformFunction can not be null");
this.jdbc = this.jdbc == null ? ConfigUtils.getJdbcConf(context.parameter().toMap(), alias) : this.jdbc;
JdbcSinkFunction<T> sinkFun = new JdbcSinkFunction<>(this.jdbc, this.sqlFunc);
return dataStream.addSink(sinkFun);
diff --git a/streampark-flink/streampark-flink-connector/streampark-flink-connector-jdbc/src/main/scala/org/apache/streampark/flink/connector/jdbc/source/JdbcJavaSource.java b/streampark-flink/streampark-flink-connector/streampark-flink-connector-jdbc/src/main/scala/org/apache/streampark/flink/connector/jdbc/source/JdbcJavaSource.java
index 42df8de68..1e19ef34c 100644
--- a/streampark-flink/streampark-flink-connector/streampark-flink-connector-jdbc/src/main/scala/org/apache/streampark/flink/connector/jdbc/source/JdbcJavaSource.java
+++ b/streampark-flink/streampark-flink-connector/streampark-flink-connector-jdbc/src/main/scala/org/apache/streampark/flink/connector/jdbc/source/JdbcJavaSource.java
@@ -53,8 +53,8 @@ public class JdbcJavaSource<T> {
SQLResultFunction<T> resultFunction,
RunningFunction runningFunc) {
- Utils.required(queryFunction != null, "queryFunction must not be null");
- Utils.required(resultFunction != null, "resultFunction must not be null");
+ Utils.notNull(queryFunction, "queryFunction must not be null");
+ Utils.notNull(resultFunction, "resultFunction must not be null");
this.jdbc = this.jdbc == null ? ConfigUtils.getJdbcConf(context.parameter().toMap(), alias) : this.jdbc;
JdbcSourceFunction<T> sourceFunction = new JdbcSourceFunction<>(jdbc, queryFunction, resultFunction, runningFunc, null);
return context.getJavaEnv().addSource(sourceFunction);
diff --git a/streampark-flink/streampark-flink-connector/streampark-flink-connector-mongo/src/main/scala/org/apache/streampark/flink/connector/mongo/source/MongoJavaSource.java b/streampark-flink/streampark-flink-connector/streampark-flink-connector-mongo/src/main/scala/org/apache/streampark/flink/connector/mongo/source/MongoJavaSource.java
index 25e63d743..261297c1c 100644
--- a/streampark-flink/streampark-flink-connector/streampark-flink-connector-mongo/src/main/scala/org/apache/streampark/flink/connector/mongo/source/MongoJavaSource.java
+++ b/streampark-flink/streampark-flink-connector/streampark-flink-connector-mongo/src/main/scala/org/apache/streampark/flink/connector/mongo/source/MongoJavaSource.java
@@ -42,9 +42,9 @@ public class MongoJavaSource<T> {
MongoResultFunction<T> resultFunction,
RunningFunction runningFunc) {
- Utils.required(collectionName != null, "collectionName must not be null");
- Utils.required(queryFunction != null, "queryFunction must not be null");
- Utils.required(resultFunction != null, "resultFunction must not be null");
+ Utils.notNull(collectionName, "collectionName must not be null");
+ Utils.notNull(queryFunction, "queryFunction must not be null");
+ Utils.notNull(resultFunction, "resultFunction must not be null");
MongoSourceFunction<T> sourceFunction = new MongoSourceFunction<>(collectionName, property, queryFunction, resultFunction, runningFunc, null);
return context.getJavaEnv().addSource(sourceFunction);