Skip to content

Commit c1e3bc3

Browse files
kipshikipshi
authored
[INLONG-4208][Manager] Merge UpdateGroupProcessForm into GroupResourceProcessForm (#4209)
1 parent 52e68e4 commit c1e3bc3

31 files changed

+106
-193
lines changed

‎inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/BaseProcessForm.java

-1
Original file line numberDiff line numberDiff line change
@@ -31,7 +31,6 @@
3131
@JsonSubTypes.Type(value = NewGroupProcessForm.class, name = NewGroupProcessForm.FORM_NAME),
3232
@JsonSubTypes.Type(value = NewConsumptionProcessForm.class, name = NewConsumptionProcessForm.FORM_NAME),
3333
@JsonSubTypes.Type(value = GroupResourceProcessForm.class, name = GroupResourceProcessForm.FORM_NAME),
34-
@JsonSubTypes.Type(value = UpdateGroupProcessForm.class, name = UpdateGroupProcessForm.FORM_NAME),
3534
@JsonSubTypes.Type(value = LightGroupResourceProcessForm.class, name = LightGroupResourceProcessForm.FORM_NAME),
3635
})
3736
public abstract class BaseProcessForm implements ProcessForm {

‎inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/GroupResourceProcessForm.java

+8-2
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,9 @@
2020
import com.google.common.collect.Maps;
2121
import lombok.Data;
2222
import lombok.EqualsAndHashCode;
23+
import lombok.Getter;
24+
import lombok.Setter;
25+
import org.apache.inlong.manager.common.enums.GroupOperateType;
2326
import org.apache.inlong.manager.common.exceptions.FormValidateException;
2427
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
2528
import org.apache.inlong.manager.common.pojo.stream.InlongStreamInfo;
@@ -38,14 +41,16 @@ public class GroupResourceProcessForm extends BaseProcessForm {
3841

3942
private InlongGroupInfo groupInfo;
4043

44+
@Getter
45+
@Setter
46+
private GroupOperateType groupOperateType = GroupOperateType.INIT;
47+
4148
@Deprecated
4249
private String streamId;
4350

4451
@Deprecated
4552
private List<InlongStreamInfo> streamInfos;
4653

47-
private List<InlongStreamInfo> streamInfoList;
48-
4954
public InlongGroupInfo getGroupInfo() {
5055
return groupInfo;
5156
}
@@ -82,6 +87,7 @@ public void setInlongStreamId(String streamId) {
8287
public Map<String, Object> showInList() {
8388
Map<String, Object> show = Maps.newHashMap();
8489
show.put("inlongGroupId", groupInfo.getInlongGroupId());
90+
show.put("groupOperateType", this.groupOperateType);
8591
return show;
8692
}
8793
}

‎inlong-manager/manager-common/src/main/java/org/apache/inlong/manager/common/pojo/workflow/form/UpdateGroupProcessForm.java

-73
This file was deleted.

‎inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/eventselect/DeleteProcessSelector.java

+6-5
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020
import lombok.SneakyThrows;
2121
import lombok.extern.slf4j.Slf4j;
2222
import org.apache.inlong.manager.common.enums.GroupOperateType;
23+
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
2324
import org.apache.inlong.manager.common.pojo.workflow.form.ProcessForm;
24-
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
2525
import org.apache.inlong.manager.workflow.WorkflowContext;
2626
import org.apache.inlong.manager.workflow.event.EventSelector;
2727

@@ -36,13 +36,14 @@ public class DeleteProcessSelector implements EventSelector {
3636
public boolean accept(WorkflowContext context) {
3737
ProcessForm processForm = context.getProcessForm();
3838
String groupId = processForm.getInlongGroupId();
39-
if (!(processForm instanceof UpdateGroupProcessForm)) {
40-
log.info("not add deleteProcess listener as the form was not UpdateGroup for groupId [{}]", groupId);
39+
if (!(processForm instanceof GroupResourceProcessForm)) {
40+
log.info("not add deleteProcess listener as the form was not GroupResourceProcessForm for groupId [{}]",
41+
groupId);
4142
return false;
4243
}
4344

44-
UpdateGroupProcessForm updateProcessForm = (UpdateGroupProcessForm) processForm;
45-
boolean flag = updateProcessForm.getGroupOperateType() == GroupOperateType.DELETE;
45+
GroupResourceProcessForm groupResourceProcessForm = (GroupResourceProcessForm) processForm;
46+
boolean flag = groupResourceProcessForm.getGroupOperateType() == GroupOperateType.DELETE;
4647
if (!flag) {
4748
log.info("not add deleteProcess listener as the operate was not DELETE for groupId [{}]", groupId);
4849
return false;

‎inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/eventselect/RestartProcessSelector.java

+5-4
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020
import lombok.SneakyThrows;
2121
import lombok.extern.slf4j.Slf4j;
2222
import org.apache.inlong.manager.common.enums.GroupOperateType;
23+
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
2324
import org.apache.inlong.manager.common.pojo.workflow.form.ProcessForm;
24-
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
2525
import org.apache.inlong.manager.workflow.WorkflowContext;
2626
import org.apache.inlong.manager.workflow.event.EventSelector;
2727

@@ -36,12 +36,13 @@ public class RestartProcessSelector implements EventSelector {
3636
public boolean accept(WorkflowContext workflowContext) {
3737
ProcessForm processForm = workflowContext.getProcessForm();
3838
String groupId = processForm.getInlongGroupId();
39-
if (!(processForm instanceof UpdateGroupProcessForm)) {
40-
log.info("not add restartProcess listener as the form was not UpdateGroup for groupId [{}]", groupId);
39+
if (!(processForm instanceof GroupResourceProcessForm)) {
40+
log.info("not add restartProcess listener as the form was not GroupResourceProcessForm for groupId [{}]",
41+
groupId);
4142
return false;
4243
}
4344

44-
UpdateGroupProcessForm updateProcessForm = (UpdateGroupProcessForm) processForm;
45+
GroupResourceProcessForm updateProcessForm = (GroupResourceProcessForm) processForm;
4546
boolean flag = updateProcessForm.getGroupOperateType() == GroupOperateType.RESTART;
4647
if (!flag) {
4748
log.info("not add restartProcess listener as the operate was not RESTART for groupId [{}]", groupId);

‎inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/eventselect/SuspendProcessSelector.java

+6-5
Original file line numberDiff line numberDiff line change
@@ -20,8 +20,8 @@
2020
import lombok.SneakyThrows;
2121
import lombok.extern.slf4j.Slf4j;
2222
import org.apache.inlong.manager.common.enums.GroupOperateType;
23+
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
2324
import org.apache.inlong.manager.common.pojo.workflow.form.ProcessForm;
24-
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
2525
import org.apache.inlong.manager.workflow.WorkflowContext;
2626
import org.apache.inlong.manager.workflow.event.EventSelector;
2727

@@ -36,13 +36,14 @@ public class SuspendProcessSelector implements EventSelector {
3636
public boolean accept(WorkflowContext workflowContext) {
3737
ProcessForm processForm = workflowContext.getProcessForm();
3838
String groupId = processForm.getInlongGroupId();
39-
if (!(processForm instanceof UpdateGroupProcessForm)) {
40-
log.info("not add suspendProcess listener as the form was not UpdateGroup for groupId [{}]", groupId);
39+
if (!(processForm instanceof GroupResourceProcessForm)) {
40+
log.info("not add suspendProcess listener as the form was not GroupResourceProcessForm for groupId [{}]",
41+
groupId);
4142
return false;
4243
}
4344

44-
UpdateGroupProcessForm updateProcessForm = (UpdateGroupProcessForm) processForm;
45-
boolean flag = updateProcessForm.getGroupOperateType() == GroupOperateType.SUSPEND;
45+
GroupResourceProcessForm groupResourceProcessForm = (GroupResourceProcessForm) processForm;
46+
boolean flag = groupResourceProcessForm.getGroupOperateType() == GroupOperateType.SUSPEND;
4647
if (!flag) {
4748
log.info("not add suspendProcess listener as the operate was not SUSPEND for groupId [{}]", groupId);
4849
return false;

‎inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/DeleteSortListener.java

+5-5
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,8 @@
2222
import org.apache.commons.lang3.StringUtils;
2323
import org.apache.inlong.manager.common.pojo.group.InlongGroupExtInfo;
2424
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
25+
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
2526
import org.apache.inlong.manager.common.pojo.workflow.form.ProcessForm;
26-
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
2727
import org.apache.inlong.manager.common.settings.InlongGroupSettings;
2828
import org.apache.inlong.manager.common.util.JsonUtils;
2929
import org.apache.inlong.manager.plugin.flink.FlinkOperation;
@@ -55,14 +55,14 @@ public TaskEvent event() {
5555
public ListenerResult listen(WorkflowContext context) throws Exception {
5656
ProcessForm processForm = context.getProcessForm();
5757
String groupId = processForm.getInlongGroupId();
58-
if (!(processForm instanceof UpdateGroupProcessForm)) {
59-
String message = String.format("process form was not UpdateGroup for groupId [%s]", groupId);
58+
if (!(processForm instanceof GroupResourceProcessForm)) {
59+
String message = String.format("process form was not GroupResourceProcessForm for groupId [%s]", groupId);
6060
log.error(message);
6161
return ListenerResult.fail(message);
6262
}
6363

64-
UpdateGroupProcessForm updateGroupForm = (UpdateGroupProcessForm) processForm;
65-
InlongGroupInfo inlongGroupInfo = updateGroupForm.getGroupInfo();
64+
GroupResourceProcessForm groupResourceProcessForm = (GroupResourceProcessForm) processForm;
65+
InlongGroupInfo inlongGroupInfo = groupResourceProcessForm.getGroupInfo();
6666
List<InlongGroupExtInfo> extList = inlongGroupInfo.getExtList();
6767
log.info("inlong group ext info: {}", extList);
6868

‎inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/RestartSortListener.java

+5-5
Original file line numberDiff line numberDiff line change
@@ -23,8 +23,8 @@
2323
import org.apache.commons.lang3.StringUtils;
2424
import org.apache.inlong.manager.common.pojo.group.InlongGroupExtInfo;
2525
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
26+
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
2627
import org.apache.inlong.manager.common.pojo.workflow.form.ProcessForm;
27-
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
2828
import org.apache.inlong.manager.common.settings.InlongGroupSettings;
2929
import org.apache.inlong.manager.common.util.JsonUtils;
3030
import org.apache.inlong.manager.plugin.flink.enums.Constants;
@@ -59,14 +59,14 @@ public TaskEvent event() {
5959
public ListenerResult listen(WorkflowContext context) throws Exception {
6060
ProcessForm processForm = context.getProcessForm();
6161
String groupId = processForm.getInlongGroupId();
62-
if (!(processForm instanceof UpdateGroupProcessForm)) {
63-
String message = String.format("process form was not UpdateGroup for groupId [%s]", groupId);
62+
if (!(processForm instanceof GroupResourceProcessForm)) {
63+
String message = String.format("process form was not GroupResourceProcessForm for groupId [%s]", groupId);
6464
log.error(message);
6565
return ListenerResult.fail(message);
6666
}
6767

68-
UpdateGroupProcessForm updateGroupForm = (UpdateGroupProcessForm) processForm;
69-
InlongGroupInfo inlongGroupInfo = updateGroupForm.getGroupInfo();
68+
GroupResourceProcessForm groupResourceProcessForm = (GroupResourceProcessForm) processForm;
69+
InlongGroupInfo inlongGroupInfo = groupResourceProcessForm.getGroupInfo();
7070
List<InlongGroupExtInfo> extList = inlongGroupInfo.getExtList();
7171
log.info("inlong group ext info: {}", extList);
7272

‎inlong-manager/manager-plugins/src/main/java/org/apache/inlong/manager/plugin/listener/SuspendSortListener.java

+5-5
Original file line numberDiff line numberDiff line change
@@ -22,8 +22,8 @@
2222
import org.apache.commons.lang3.StringUtils;
2323
import org.apache.inlong.manager.common.pojo.group.InlongGroupExtInfo;
2424
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
25+
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
2526
import org.apache.inlong.manager.common.pojo.workflow.form.ProcessForm;
26-
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
2727
import org.apache.inlong.manager.common.settings.InlongGroupSettings;
2828
import org.apache.inlong.manager.common.util.JsonUtils;
2929
import org.apache.inlong.manager.plugin.flink.FlinkOperation;
@@ -55,14 +55,14 @@ public TaskEvent event() {
5555
public ListenerResult listen(WorkflowContext context) throws Exception {
5656
ProcessForm processForm = context.getProcessForm();
5757
String groupId = processForm.getInlongGroupId();
58-
if (!(processForm instanceof UpdateGroupProcessForm)) {
59-
String message = String.format("process form was not UpdateGroup for groupId [%s]", groupId);
58+
if (!(processForm instanceof GroupResourceProcessForm)) {
59+
String message = String.format("process form was not GroupResourceProcessForm for groupId [%s]", groupId);
6060
log.error(message);
6161
return ListenerResult.fail(message);
6262
}
6363

64-
UpdateGroupProcessForm updateGroupForm = (UpdateGroupProcessForm) processForm;
65-
InlongGroupInfo inlongGroupInfo = updateGroupForm.getGroupInfo();
64+
GroupResourceProcessForm groupResourceProcessForm = (GroupResourceProcessForm) processForm;
65+
InlongGroupInfo inlongGroupInfo = groupResourceProcessForm.getGroupInfo();
6666
List<InlongGroupExtInfo> extList = inlongGroupInfo.getExtList();
6767
log.info("inlong group ext info: {}", extList);
6868

‎inlong-manager/manager-plugins/src/test/java/org/apache/inlong/manager/plugin/listener/DeleteSortListenerTest.java

+4-4
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
import com.fasterxml.jackson.databind.ObjectMapper;
2121
import org.apache.inlong.manager.common.pojo.group.InlongGroupExtInfo;
2222
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
23-
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
23+
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
2424
import org.apache.inlong.manager.common.settings.InlongGroupSettings;
2525
import org.apache.inlong.manager.plugin.flink.enums.Constants;
2626
import org.apache.inlong.manager.workflow.WorkflowContext;
@@ -39,11 +39,11 @@ public class DeleteSortListenerTest {
3939
@Test
4040
public void testListener() throws Exception {
4141
WorkflowContext context = new WorkflowContext();
42-
UpdateGroupProcessForm updateGroupProcessForm = new UpdateGroupProcessForm();
43-
context.setProcessForm(updateGroupProcessForm);
42+
GroupResourceProcessForm groupResourceProcessForm = new GroupResourceProcessForm();
43+
context.setProcessForm(groupResourceProcessForm);
4444
InlongGroupInfo inlongGroupInfo = new InlongGroupInfo();
4545
inlongGroupInfo.setInlongGroupId("1");
46-
updateGroupProcessForm.setGroupInfo(inlongGroupInfo);
46+
groupResourceProcessForm.setGroupInfo(inlongGroupInfo);
4747

4848
InlongGroupExtInfo inlongGroupExtInfo1 = new InlongGroupExtInfo();
4949
inlongGroupExtInfo1.setKeyName(InlongGroupSettings.SORT_URL);

‎inlong-manager/manager-plugins/src/test/java/org/apache/inlong/manager/plugin/listener/RestartSortListenerTest.java

+4-4
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
import com.fasterxml.jackson.databind.ObjectMapper;
2121
import org.apache.inlong.manager.common.pojo.group.InlongGroupExtInfo;
2222
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
23-
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
23+
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
2424
import org.apache.inlong.manager.common.settings.InlongGroupSettings;
2525
import org.apache.inlong.manager.workflow.WorkflowContext;
2626
import org.junit.Test;
@@ -38,11 +38,11 @@ public class RestartSortListenerTest {
3838
@Test
3939
public void testListener() throws Exception {
4040
WorkflowContext context = new WorkflowContext();
41-
UpdateGroupProcessForm updateGroupProcessForm = new UpdateGroupProcessForm();
42-
context.setProcessForm(updateGroupProcessForm);
41+
GroupResourceProcessForm groupResourceProcessForm = new GroupResourceProcessForm();
42+
context.setProcessForm(groupResourceProcessForm);
4343
InlongGroupInfo inlongGroupInfo = new InlongGroupInfo();
4444
inlongGroupInfo.setInlongGroupId("1");
45-
updateGroupProcessForm.setGroupInfo(inlongGroupInfo);
45+
groupResourceProcessForm.setGroupInfo(inlongGroupInfo);
4646

4747
InlongGroupExtInfo inlongGroupExtInfo1 = new InlongGroupExtInfo();
4848
inlongGroupExtInfo1.setKeyName(InlongGroupSettings.SORT_URL);

‎inlong-manager/manager-plugins/src/test/java/org/apache/inlong/manager/plugin/listener/SuspendSortListenerTest.java

+4-4
Original file line numberDiff line numberDiff line change
@@ -20,7 +20,7 @@
2020
import com.fasterxml.jackson.databind.ObjectMapper;
2121
import org.apache.inlong.manager.common.pojo.group.InlongGroupExtInfo;
2222
import org.apache.inlong.manager.common.pojo.group.InlongGroupInfo;
23-
import org.apache.inlong.manager.common.pojo.workflow.form.UpdateGroupProcessForm;
23+
import org.apache.inlong.manager.common.pojo.workflow.form.GroupResourceProcessForm;
2424
import org.apache.inlong.manager.common.settings.InlongGroupSettings;
2525
import org.apache.inlong.manager.workflow.WorkflowContext;
2626
import org.junit.Test;
@@ -38,11 +38,11 @@ public class SuspendSortListenerTest {
3838
@Test
3939
public void testListener() throws Exception {
4040
WorkflowContext context = new WorkflowContext();
41-
UpdateGroupProcessForm updateGroupProcessForm = new UpdateGroupProcessForm();
42-
context.setProcessForm(updateGroupProcessForm);
41+
GroupResourceProcessForm groupResourceProcessForm = new GroupResourceProcessForm();
42+
context.setProcessForm(groupResourceProcessForm);
4343
InlongGroupInfo inlongGroupInfo = new InlongGroupInfo();
4444
inlongGroupInfo.setInlongGroupId("1");
45-
updateGroupProcessForm.setGroupInfo(inlongGroupInfo);
45+
groupResourceProcessForm.setGroupInfo(inlongGroupInfo);
4646

4747
InlongGroupExtInfo inlongGroupExtInfo1 = new InlongGroupExtInfo();
4848
inlongGroupExtInfo1.setKeyName(InlongGroupSettings.SORT_URL);

0 commit comments

Comments
 (0)