You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampark.apache.org by kr...@apache.org on 2023/04/02 09:49:40 UTC
[incubator-streampark] branch dev updated: FlinkHome check improvement (#2548)
This is an automated email from the ASF dual-hosted git repository.
kriszu 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 3f0031893 FlinkHome check improvement (#2548)
3f0031893 is described below
commit 3f00318937c3cc5396478fed8713eaf93274a1e2
Author: benjobs <be...@apache.org>
AuthorDate: Sun Apr 2 17:49:33 2023 +0800
FlinkHome check improvement (#2548)
---
.../core/controller/FlinkEnvController.java | 8 +-
.../console/core/service/FlinkEnvService.java | 3 +-
.../core/service/impl/FlinkEnvServiceImpl.java | 28 ++++-
.../src/api/flink/setting/flinkEnv.ts | 6 +-
.../views/setting/FlinkHome/components/Modal.vue | 129 +++++++++------------
5 files changed, 90 insertions(+), 84 deletions(-)
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkEnvController.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkEnvController.java
index 7c2cc7b92..4686dccd2 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkEnvController.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/controller/FlinkEnvController.java
@@ -45,10 +45,10 @@ public class FlinkEnvController {
return RestResponse.success(list);
}
- @PostMapping("exists")
- public RestResponse exists(FlinkEnv version) {
- boolean checked = flinkEnvService.exists(version);
- return RestResponse.success(checked);
+ @PostMapping("check")
+ public RestResponse check(FlinkEnv version) {
+ Integer checkResp = flinkEnvService.check(version);
+ return RestResponse.success(checkResp);
}
@PostMapping("create")
diff --git a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/FlinkEnvService.java b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/FlinkEnvService.java
index 3ad42bc39..8336f1255 100644
--- a/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/FlinkEnvService.java
+++ b/streampark-console/streampark-console-service/src/main/java/org/apache/streampark/console/core/service/FlinkEnvService.java
@@ -31,8 +31,7 @@ public interface FlinkEnvService extends IService<FlinkEnv> {
* @param version
* @return
*/
- boolean exists(FlinkEnv version);
-
+ Integer check(FlinkEnv version);
/**
* create new
*
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 f874c1d40..0495ac421 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
@@ -29,6 +29,7 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
+import java.io.File;
import java.io.IOException;
import java.util.Date;
@@ -38,16 +39,37 @@ import java.util.Date;
public class FlinkEnvServiceImpl extends ServiceImpl<FlinkEnvMapper, FlinkEnv>
implements FlinkEnvService {
- /** two places will be checked: 1) name cannot be repeated 2) flink version need to be parsed */
+ /**
+ * two places will be checked: <br>
+ * 1) name repeated <br>
+ * 2) flink-dist <br>
+ * -1) invalid path <br>
+ * 0) ok <br>
+ */
@Override
- public boolean exists(FlinkEnv version) {
+ public Integer check(FlinkEnv version) {
// 1) check name
LambdaQueryWrapper<FlinkEnv> queryWrapper =
new LambdaQueryWrapper<FlinkEnv>().eq(FlinkEnv::getFlinkName, version.getFlinkName());
if (version.getId() != null) {
queryWrapper.ne(FlinkEnv::getId, version.getId());
}
- return this.count(queryWrapper) == 0;
+ if (this.count(queryWrapper) > 0) {
+ return 1;
+ }
+
+ // 2) check dist_jar
+ String lib = version.getFlinkHome().concat("/lib");
+ File flinkLib = new File(lib);
+ if (flinkLib.exists() && flinkLib.isDirectory()) {
+ int distSize = flinkLib.listFiles(f -> f.getName().matches("flink-dist.*\\.jar")).length;
+ if (distSize > 1) {
+ return 2;
+ }
+ } else {
+ return -1;
+ }
+ return 0;
}
@Override
diff --git a/streampark-console/streampark-console-webapp/src/api/flink/setting/flinkEnv.ts b/streampark-console/streampark-console-webapp/src/api/flink/setting/flinkEnv.ts
index 312e4a5bf..84a0911eb 100644
--- a/streampark-console/streampark-console-webapp/src/api/flink/setting/flinkEnv.ts
+++ b/streampark-console/streampark-console-webapp/src/api/flink/setting/flinkEnv.ts
@@ -22,7 +22,7 @@ import { defHttp } from '/@/utils/http/axios';
enum FLINK_API {
LIST = '/flink/env/list',
CREATE = '/flink/env/create',
- EXISTS = '/flink/env/exists',
+ CHECK = '/flink/env/check',
GET = '/flink/env/get',
SYNC = '/flink/env/sync',
UPDATE = '/flink/env/update',
@@ -66,12 +66,12 @@ export function fetchFlinkInfo(id: string): Promise<FlinkEnv> {
* @param {Recordable} data
* @returns {Promise<Boolean>}
*/
-export function fetchExistsEnv(data: {
+export function fetchCheckEnv(data: {
id: string | null;
flinkName: string;
flinkHome: string;
}): Promise<AxiosResponse<Result<boolean>>> {
- return defHttp.post({ url: FLINK_API.EXISTS, data }, { isReturnNativeResponse: true });
+ return defHttp.post({ url: FLINK_API.CHECK, data }, { isReturnNativeResponse: true });
}
/**
* Create flink
diff --git a/streampark-console/streampark-console-webapp/src/views/setting/FlinkHome/components/Modal.vue b/streampark-console/streampark-console-webapp/src/views/setting/FlinkHome/components/Modal.vue
index 5dc9010e7..5cad2ed23 100644
--- a/streampark-console/streampark-console-webapp/src/views/setting/FlinkHome/components/Modal.vue
+++ b/streampark-console/streampark-console-webapp/src/views/setting/FlinkHome/components/Modal.vue
@@ -28,16 +28,12 @@
import { SvgIcon } from '/@/components/Icon';
import { BasicModal, useModalInner } from '/@/components/Modal';
import { useMessage } from '/@/hooks/web/useMessage';
- import {
- fetchExistsEnv,
- fetchFlinkCreate,
- fetchFlinkUpdate,
- } from '/@/api/flink/setting/flinkEnv';
+ import { fetchCheckEnv, fetchFlinkCreate, fetchFlinkUpdate } from '/@/api/flink/setting/flinkEnv';
const emit = defineEmits(['reload', 'register']);
const versionId = ref<string | null>(null);
const { t } = useI18n();
- const { createConfirm, Swal, createMessage } = useMessage();
+ const { Swal } = useMessage();
const [registerForm, { setFieldsValue, validate, resetFields }] = useForm({
labelWidth: 120,
colon: true,
@@ -95,82 +91,71 @@
/* form submit */
async function handleSubmit() {
- try {
- changeOkLoading(true);
- const formValue = await validate();
- // Detection environment
- const { data: resp } = await fetchExistsEnv({
- id: versionId.value,
- flinkName: formValue.flinkName,
- flinkHome: formValue.flinkHome,
- });
+ changeOkLoading(true);
+ const formValue = await validate();
+ // Detection environment
+ const { data: resp } = await fetchCheckEnv({
+ id: versionId.value,
+ flinkName: formValue.flinkName,
+ flinkHome: formValue.flinkHome,
+ });
+ const checkResp = parseInt(resp.data);
+ if (checkResp != 0) {
// Environment detection is successful
- if (resp.data) {
- let message: string;
- let success = false;
- // create
- if (versionId.value == null) {
- const { data } = await fetchFlinkCreate(formValue);
- if (data.data) {
- success = true;
- message = formValue.flinkName.concat(
- t('setting.flinkHome.operateMessage.createFlinkHomeSuccessful'),
- );
- } else {
- message = data.message;
- }
- } else {
- // update
- const { data } = await fetchFlinkUpdate({
- id: versionId.value,
- ...formValue,
- });
- if (data.data) {
- message = formValue.flinkName.concat(
- t('setting.flinkHome.operateMessage.updateFlinkHomeSuccessful'),
- );
- success = true;
- } else {
- message = data.message;
- }
- }
- if (success) {
- Swal.fire({
- icon: 'success',
- title: message,
- showConfirmButton: false,
- timer: 2000,
- });
- closeModal();
- emit('reload');
+ if (checkResp == -1) {
+ Swal.fire('Failed', 'FLINK_HOME invalid path.', 'error');
+ } else if (checkResp == 1) {
+ Swal.fire('Failed', t('setting.flinkHome.operateMessage.flinkNameIsUnique'), 'error');
+ } else if (checkResp == 2) {
+ Swal.fire(
+ 'Failed',
+ 'can no found flink-dist or found multiple flink-dist, FLINK_HOME error.',
+ 'error',
+ );
+ }
+ return;
+ }
+
+ try {
+ let message: string;
+ let success = false;
+ // create
+ if (versionId.value == null) {
+ const { data } = await fetchFlinkCreate(formValue);
+ if (data.data) {
+ success = true;
+ message = formValue.flinkName.concat(
+ t('setting.flinkHome.operateMessage.createFlinkHomeSuccessful'),
+ );
} else {
- Swal.fire('Failed', message.replaceAll(/\[StreamPark]/g, ''), 'error');
+ message = data.message;
}
} else {
- if (resp.status === 'error') {
- Swal.fire(
- 'Failed',
- 'can no found flink-dist or found multiple flink-dist, FLINK_HOME error.',
- 'error',
+ // update
+ const { data } = await fetchFlinkUpdate({
+ id: versionId.value,
+ ...formValue,
+ });
+ if (data.data) {
+ message = formValue.flinkName.concat(
+ t('setting.flinkHome.operateMessage.updateFlinkHomeSuccessful'),
);
+ success = true;
} else {
- Swal.fire('Failed', t('setting.flinkHome.operateMessage.flinkNameIsUnique'), 'error');
+ message = data.message;
}
}
- } catch (error: any) {
- /* custom alert message */
- if (error?.response?.data?.message) {
- createConfirm({
- iconType: 'error',
- title: 'Operation Failed',
- content: h(
- 'div',
- { class: 'whitespace-pre-wrap' },
- error?.response?.data?.message.replaceAll(/\[StreamPark]/g, ''),
- ),
+ if (success) {
+ Swal.fire({
+ icon: 'success',
+ title: message,
+ showConfirmButton: false,
+ timer: 2000,
});
+ closeModal();
+ emit('reload');
} else {
- createMessage.error('error');
+ Swal.fire('Failed', message.replaceAll(/\[StreamPark]/g, ''), 'error');
}
} finally {
changeOkLoading(false);