From 623941bb6d34336e6d1f22ca55244bd90a36109a Mon Sep 17 00:00:00 2001
From: qianmoQ <shicheng@ttxit.com>
Date: Fri, 10 Mar 2023 11:25:29 +0800
Subject: [PATCH 1/5] [Web] Support ChatGPT multi-information recording

---
 core/datacap-web/console-fe/src/model/User.ts |  6 +++
 .../views/pages/profile/ProfileChatGPT.vue    | 51 ++++++++++++++++---
 2 files changed, 51 insertions(+), 6 deletions(-)

diff --git a/core/datacap-web/console-fe/src/model/User.ts b/core/datacap-web/console-fe/src/model/User.ts
index c8443af8..5c27dcb5 100644
--- a/core/datacap-web/console-fe/src/model/User.ts
+++ b/core/datacap-web/console-fe/src/model/User.ts
@@ -11,6 +11,12 @@ export class UserQuestion
   type: string;
 }
 
+export class UserQuestionItem
+{
+  content: string;
+  isSelf: boolean;
+}
+
 export class ThirdConfigure
 {
   token = '';
diff --git a/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue b/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue
index 4225defe..114976db 100644
--- a/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue
+++ b/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue
@@ -12,9 +12,23 @@
       <div>
         <Layout>
           <Content style="padding: 24px 50px;">
-            <Input v-model="userQuestionResponse" :disabled="!userInfo?.thirdConfigure?.chatgpt?.token" type="textarea" :rows="10"/>
+            <div ref="scrollDiv" style="height: 300px; max-height: 300px; overflow: auto; background-color: #f5f7f9">
+              <List item-layout="vertical">
+                <ListItem v-for="item in userQuestionItems" :key="item">
+                  <ListItemMeta :description="item.content">
+                    <template #title>
+                      {{ item.isSelf ? username : 'ChatGPT'}}
+                    </template>
+                    <template #avatar>
+                      <Avatar v-if="item.isSelf" icon="ios-person"></Avatar>
+                      <Avatar v-else icon="md-ionitron" style="background-color: #87d068;"></Avatar>
+                    </template>
+                  </ListItemMeta>
+                </ListItem>
+              </List>
+            </div>
           </Content>
-          <Footer>
+          <Footer style="background-color: #FFFFFF;">
             <Row>
               <Col span="20">
                 <Input v-model="userQuestion.question" :disabled="!userInfo?.thirdConfigure?.chatgpt?.token" type="textarea" :autosize="{minRows: 2,maxRows: 5}"/>
@@ -50,15 +64,23 @@
 <script lang="ts">
 import {defineComponent} from 'vue';
 import UserService from "@/services/UserService";
-import {ThirdConfigure, User, UserQuestion} from '@/model/User';
+import {ThirdConfigure, User, UserQuestion, UserQuestionItem} from '@/model/User';
+import Common from "@/common/Common";
+import {AuthResponse} from "@/model/AuthResponse";
 
 const userQuestion = new UserQuestion();
 userQuestion.type = 'ChatGPT';
 export default defineComponent({
   setup()
   {
+    let username;
+    const authUser = JSON.parse(localStorage.getItem(Common.token) || '{}') as AuthResponse;
+    if (authUser) {
+      username = authUser.username;
+    }
     return {
-      userQuestion
+      userQuestion,
+      username
     }
   },
   created()
@@ -72,12 +94,13 @@ export default defineComponent({
       visibleModel: false,
       startChatLoading: false,
       userInfo: null as User,
-      userQuestionResponse: null
+      userQuestionItems: null as UserQuestionItem[]
     }
   },
   methods: {
     handlerInitialize()
     {
+      this.userQuestionItems = [];
       this.loading = true;
       UserService.getInfo()
         .then(response => {
@@ -118,11 +141,20 @@ export default defineComponent({
     },
     handlerStartChat()
     {
+      const question = new UserQuestionItem();
+      question.content = this.userQuestion.question;
+      question.isSelf = true;
+      this.userQuestionItems.push(question);
+      this.handlerGoBottom();
       this.startChatLoading = true;
       UserService.startChat(this.userQuestion)
         .then(response => {
           if (response.status) {
-            this.userQuestionResponse = response.data;
+            const answer = new UserQuestionItem();
+            answer.content = response.data.toString();
+            answer.isSelf = false;
+            this.userQuestionItems.push(answer);
+            this.handlerGoBottom();
           }
           else {
             this.$Message.error(response.message);
@@ -131,6 +163,13 @@ export default defineComponent({
         .finally(() => {
           this.startChatLoading = false;
         });
+    },
+    handlerGoBottom()
+    {
+      let scrollElem = this.$refs.scrollDiv;
+      setTimeout(() => {
+        scrollElem.scrollTo({top: scrollElem.scrollHeight, behavior: 'smooth'});
+      }, 0);
     }
   }
 });

From 2f692ceae7e66e36035fcb150d8f884e8bcee60c Mon Sep 17 00:00:00 2001
From: qianmoQ <shicheng@ttxit.com>
Date: Sat, 11 Mar 2023 10:30:49 +0800
Subject: [PATCH 2/5] [Web] Support ChatGPT to empty the input content after
 submission

---
 .../src/views/pages/profile/ProfileChatGPT.vue   | 16 +++++++++-------
 1 file changed, 9 insertions(+), 7 deletions(-)

diff --git a/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue b/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue
index 114976db..0dd64891 100644
--- a/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue
+++ b/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue
@@ -17,7 +17,7 @@
                 <ListItem v-for="item in userQuestionItems" :key="item">
                   <ListItemMeta :description="item.content">
                     <template #title>
-                      {{ item.isSelf ? username : 'ChatGPT'}}
+                      {{ item.isSelf ? username : 'ChatGPT' }}
                     </template>
                     <template #avatar>
                       <Avatar v-if="item.isSelf" icon="ios-person"></Avatar>
@@ -31,7 +31,7 @@
           <Footer style="background-color: #FFFFFF;">
             <Row>
               <Col span="20">
-                <Input v-model="userQuestion.question" :disabled="!userInfo?.thirdConfigure?.chatgpt?.token" type="textarea" :autosize="{minRows: 2,maxRows: 5}"/>
+                <Input v-model="userQuestionContext" :disabled="!userInfo?.thirdConfigure?.chatgpt?.token" type="textarea" :autosize="{minRows: 2,maxRows: 5}"/>
               </Col>
               <Col span="2" offset="1">
                 <Button :disabled="!userInfo?.thirdConfigure?.chatgpt?.token" type="primary" icon="md-send"
@@ -68,8 +68,6 @@ import {ThirdConfigure, User, UserQuestion, UserQuestionItem} from '@/model/User
 import Common from "@/common/Common";
 import {AuthResponse} from "@/model/AuthResponse";
 
-const userQuestion = new UserQuestion();
-userQuestion.type = 'ChatGPT';
 export default defineComponent({
   setup()
   {
@@ -79,7 +77,6 @@ export default defineComponent({
       username = authUser.username;
     }
     return {
-      userQuestion,
       username
     }
   },
@@ -94,6 +91,7 @@ export default defineComponent({
       visibleModel: false,
       startChatLoading: false,
       userInfo: null as User,
+      userQuestionContext: null,
       userQuestionItems: null as UserQuestionItem[]
     }
   },
@@ -141,13 +139,17 @@ export default defineComponent({
     },
     handlerStartChat()
     {
+      const userQuestion = new UserQuestion();
+      userQuestion.type = 'ChatGPT';
+      userQuestion.question = this.userQuestionContext;
       const question = new UserQuestionItem();
-      question.content = this.userQuestion.question;
+      question.content = this.userQuestionContext;
       question.isSelf = true;
       this.userQuestionItems.push(question);
       this.handlerGoBottom();
       this.startChatLoading = true;
-      UserService.startChat(this.userQuestion)
+      this.userQuestionContext = null;
+      UserService.startChat(userQuestion)
         .then(response => {
           if (response.status) {
             const answer = new UserQuestionItem();

From 006d6ce50359f4e0a691f012bfee68ad380f6fda Mon Sep 17 00:00:00 2001
From: qianmoQ <shicheng@ttxit.com>
Date: Sat, 11 Mar 2023 11:04:03 +0800
Subject: [PATCH 3/5] [Web] Support ChatGPT format context

---
 .../console-fe/src/views/pages/profile/ProfileChatGPT.vue    | 5 +++--
 1 file changed, 3 insertions(+), 2 deletions(-)

diff --git a/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue b/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue
index 0dd64891..d3a93195 100644
--- a/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue
+++ b/core/datacap-web/console-fe/src/views/pages/profile/ProfileChatGPT.vue
@@ -11,11 +11,11 @@
       </template>
       <div>
         <Layout>
-          <Content style="padding: 24px 50px;">
+          <Content style="padding: 0px 0px 0px 10px">
             <div ref="scrollDiv" style="height: 300px; max-height: 300px; overflow: auto; background-color: #f5f7f9">
               <List item-layout="vertical">
                 <ListItem v-for="item in userQuestionItems" :key="item">
-                  <ListItemMeta :description="item.content">
+                  <ListItemMeta style="margin-bottom: 0px;">
                     <template #title>
                       {{ item.isSelf ? username : 'ChatGPT' }}
                     </template>
@@ -24,6 +24,7 @@
                       <Avatar v-else icon="md-ionitron" style="background-color: #87d068;"></Avatar>
                     </template>
                   </ListItemMeta>
+                  <v-md-preview :text="item.content"/>
                 </ListItem>
               </List>
             </div>

From be583b1a5c52dbb3ddc58d5e75fcaa6283b2ede5 Mon Sep 17 00:00:00 2001
From: qianmoQ <shicheng@ttxit.com>
Date: Sat, 11 Mar 2023 11:47:08 +0800
Subject: [PATCH 4/5] [Web] Add system announcement display

---
 core/datacap-web/console-fe/src/views/layout/Layout.vue | 5 +++++
 1 file changed, 5 insertions(+)

diff --git a/core/datacap-web/console-fe/src/views/layout/Layout.vue b/core/datacap-web/console-fe/src/views/layout/Layout.vue
index 90f27880..c42c71aa 100644
--- a/core/datacap-web/console-fe/src/views/layout/Layout.vue
+++ b/core/datacap-web/console-fe/src/views/layout/Layout.vue
@@ -1,6 +1,11 @@
 <template>
   <div class="layout">
     <Layout>
+      <RouterLink to="/profile/chatgpt">
+        <Alert style="margin-bottom: 0px;" banner closable type="success">
+          Support ChatGPT
+        </Alert>
+      </RouterLink>
       <LayoutHeader @changeLanguage="setLangCondition($event)"/>
       <LayoutContent style="background-color: #FFFFFF; padding: 12px; min-height: 500px"/>
       <LayoutFooter/>

From 6c795141c909984863e1987c315d4746de85273e Mon Sep 17 00:00:00 2001
From: qianmoQ <shicheng@ttxit.com>
Date: Sun, 12 Mar 2023 17:56:52 +0800
Subject: [PATCH 5/5] [Core] [Executor] Add submit pipeline api

---
 .../src/main/etc/conf/application.properties  |   2 +
 .../etc/conf/plugins/jdbc/clickhouse.json     |  28 +++++
 .../datacap/server/body/PipelineBody.java     |  18 +++
 .../server/common/BeanToPropertiesCommon.java |  34 ++++++
 .../controller/user/PipelineController.java   |  26 +++++
 .../server/plugin/configure/IConfigure.java   |   1 +
 .../plugin/configure/IConfigureExecutor.java  |  19 ++++
 .../configure/IConfigureExecutorField.java    |  16 +++
 .../server/service/PipelineService.java       |   9 ++
 .../service/impl/PipelineServiceImpl.java     | 103 ++++++++++++++++++
 10 files changed, 256 insertions(+)
 create mode 100644 core/datacap-server/src/main/java/io/edurt/datacap/server/body/PipelineBody.java
 create mode 100644 core/datacap-server/src/main/java/io/edurt/datacap/server/common/BeanToPropertiesCommon.java
 create mode 100644 core/datacap-server/src/main/java/io/edurt/datacap/server/controller/user/PipelineController.java
 create mode 100644 core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigureExecutor.java
 create mode 100644 core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigureExecutorField.java
 create mode 100644 core/datacap-server/src/main/java/io/edurt/datacap/server/service/PipelineService.java
 create mode 100644 core/datacap-server/src/main/java/io/edurt/datacap/server/service/impl/PipelineServiceImpl.java

diff --git a/core/datacap-server/src/main/etc/conf/application.properties b/core/datacap-server/src/main/etc/conf/application.properties
index 465f8fa7..b17b9f2a 100644
--- a/core/datacap-server/src/main/etc/conf/application.properties
+++ b/core/datacap-server/src/main/etc/conf/application.properties
@@ -9,3 +9,5 @@ spring.datasource.username=root
 spring.datasource.password=12345678
 datacap.security.secret=DataCapSecretKey
 datacap.security.expiration=86400000
+# Executor
+datacap.executor.seatunnel.home=/opt/lib/seatunnel
diff --git a/core/datacap-server/src/main/etc/conf/plugins/jdbc/clickhouse.json b/core/datacap-server/src/main/etc/conf/plugins/jdbc/clickhouse.json
index 013d4f01..877da9f9 100644
--- a/core/datacap-server/src/main/etc/conf/plugins/jdbc/clickhouse.json
+++ b/core/datacap-server/src/main/etc/conf/plugins/jdbc/clickhouse.json
@@ -45,5 +45,33 @@
       "value": [],
       "group": "custom"
     }
+  ],
+  "pipelines": [
+    {
+      "executor": "Seatunnel",
+      "type": "FROM",
+      "fields": [
+        {
+          "field": "host",
+          "required": true
+        },
+        {
+          "field": "database",
+          "required": true
+        },
+        {
+          "field": "sql",
+          "required": true
+        },
+        {
+          "field": "username",
+          "required": true
+        },
+        {
+          "field": "password",
+          "required": true
+        }
+      ]
+    }
   ]
 }
diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/body/PipelineBody.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/body/PipelineBody.java
new file mode 100644
index 00000000..0b8ad320
--- /dev/null
+++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/body/PipelineBody.java
@@ -0,0 +1,18 @@
+package io.edurt.datacap.server.body;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+@Data
+@ToString
+@NoArgsConstructor
+@AllArgsConstructor
+public class PipelineBody
+{
+    private Long from;
+    private Long to;
+    private String content;
+    private String executor;
+}
diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/common/BeanToPropertiesCommon.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/common/BeanToPropertiesCommon.java
new file mode 100644
index 00000000..8b720b3c
--- /dev/null
+++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/common/BeanToPropertiesCommon.java
@@ -0,0 +1,34 @@
+package io.edurt.datacap.server.common;
+
+import java.beans.BeanInfo;
+import java.beans.Introspector;
+import java.beans.PropertyDescriptor;
+import java.lang.reflect.Method;
+import java.util.Properties;
+
+public class BeanToPropertiesCommon
+{
+    private BeanToPropertiesCommon()
+    {}
+
+    public static Properties convertBeanToProperties(Object bean)
+    {
+        Properties properties = new Properties();
+        try {
+            BeanInfo beanInfo = Introspector.getBeanInfo(bean.getClass());
+            PropertyDescriptor[] propertyDescriptors = beanInfo.getPropertyDescriptors();
+            for (PropertyDescriptor propertyDescriptor : propertyDescriptors) {
+                String propertyName = propertyDescriptor.getName();
+                if (!propertyName.equals("class")) {
+                    Method readMethod = propertyDescriptor.getReadMethod();
+                    Object value = readMethod.invoke(bean);
+                    properties.setProperty(propertyName, value.toString());
+                }
+            }
+        }
+        catch (Exception exception) {
+            exception.printStackTrace();
+        }
+        return properties;
+    }
+}
diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/user/PipelineController.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/user/PipelineController.java
new file mode 100644
index 00000000..ede858f0
--- /dev/null
+++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/controller/user/PipelineController.java
@@ -0,0 +1,26 @@
+package io.edurt.datacap.server.controller.user;
+
+import io.edurt.datacap.server.body.PipelineBody;
+import io.edurt.datacap.server.common.Response;
+import io.edurt.datacap.server.service.PipelineService;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController()
+@RequestMapping(value = "/api/v1/pipeline")
+public class PipelineController
+{
+    private final PipelineService pipelineService;
+
+    public PipelineController(PipelineService pipelineService)
+    {
+        this.pipelineService = pipelineService;
+    }
+
+    @PostMapping(value = "/create")
+    public Response<Object> create(PipelineBody configure)
+    {
+        return pipelineService.submit(configure);
+    }
+}
diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigure.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigure.java
index a38c190c..fdb100e8 100644
--- a/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigure.java
+++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigure.java
@@ -20,4 +20,5 @@ public class IConfigure
     private String name;
     private Date supportTime;
     private List<IConfigureField> configures;
+    private List<IConfigureExecutor> pipelines;
 }
diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigureExecutor.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigureExecutor.java
new file mode 100644
index 00000000..d8636eab
--- /dev/null
+++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigureExecutor.java
@@ -0,0 +1,19 @@
+package io.edurt.datacap.server.plugin.configure;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+import java.util.List;
+
+@Data
+@ToString
+@NoArgsConstructor
+@AllArgsConstructor
+public class IConfigureExecutor
+{
+    private String executor;
+    private String type;
+    private List<IConfigureExecutorField> fields;
+}
diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigureExecutorField.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigureExecutorField.java
new file mode 100644
index 00000000..5512bfb4
--- /dev/null
+++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/plugin/configure/IConfigureExecutorField.java
@@ -0,0 +1,16 @@
+package io.edurt.datacap.server.plugin.configure;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.ToString;
+
+@Data
+@ToString
+@NoArgsConstructor
+@AllArgsConstructor
+public class IConfigureExecutorField
+{
+    private String field;
+    private boolean required;
+}
diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/service/PipelineService.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/service/PipelineService.java
new file mode 100644
index 00000000..3e1ba50e
--- /dev/null
+++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/service/PipelineService.java
@@ -0,0 +1,9 @@
+package io.edurt.datacap.server.service;
+
+import io.edurt.datacap.server.body.PipelineBody;
+import io.edurt.datacap.server.common.Response;
+
+public interface PipelineService
+{
+    Response<Object> submit(PipelineBody configure);
+}
diff --git a/core/datacap-server/src/main/java/io/edurt/datacap/server/service/impl/PipelineServiceImpl.java b/core/datacap-server/src/main/java/io/edurt/datacap/server/service/impl/PipelineServiceImpl.java
new file mode 100644
index 00000000..4f2e289f
--- /dev/null
+++ b/core/datacap-server/src/main/java/io/edurt/datacap/server/service/impl/PipelineServiceImpl.java
@@ -0,0 +1,103 @@
+package io.edurt.datacap.server.service.impl;
+
+import com.google.inject.Injector;
+import com.google.inject.Key;
+import com.google.inject.TypeLiteral;
+import io.edurt.datacap.server.body.PipelineBody;
+import io.edurt.datacap.server.common.BeanToPropertiesCommon;
+import io.edurt.datacap.server.common.PluginCommon;
+import io.edurt.datacap.server.common.Response;
+import io.edurt.datacap.server.common.ServiceState;
+import io.edurt.datacap.server.entity.SourceEntity;
+import io.edurt.datacap.server.plugin.configure.IConfigure;
+import io.edurt.datacap.server.plugin.configure.IConfigureExecutor;
+import io.edurt.datacap.server.repository.SourceRepository;
+import io.edurt.datacap.server.service.PipelineService;
+import io.edurt.datacap.spi.executor.Executor;
+import io.edurt.datacap.spi.executor.Pipeline;
+import io.edurt.datacap.spi.executor.PipelineField;
+import org.springframework.core.env.Environment;
+import org.springframework.stereotype.Service;
+
+import java.util.Optional;
+import java.util.Properties;
+import java.util.Set;
+import java.util.UUID;
+
+@Service
+public class PipelineServiceImpl
+        implements PipelineService
+{
+    private final SourceRepository repository;
+    private final Injector injector;
+    private final Environment environment;
+
+    public PipelineServiceImpl(SourceRepository repository, Injector injector, Environment environment)
+    {
+        this.repository = repository;
+        this.injector = injector;
+        this.environment = environment;
+    }
+
+    @Override
+    public Response<Object> submit(PipelineBody configure)
+    {
+        Optional<SourceEntity> fromSourceOptional = repository.findById(configure.getFrom());
+        Optional<SourceEntity> toSourceOptional = repository.findById(configure.getTo());
+
+        if (fromSourceOptional.isPresent() && toSourceOptional.isPresent()) {
+            SourceEntity fromSource = fromSourceOptional.get();
+            IConfigure fromConfigure = PluginCommon.loadConfigure(fromSource.getProtocol(), fromSource.getType(), fromSource.getType(), environment);
+            Optional<IConfigureExecutor> fromConfigureExecutor = fromConfigure.getPipelines()
+                    .stream()
+                    .filter(v -> v.getExecutor().equals(configure.getExecutor()))
+                    .findFirst();
+            if (!fromConfigureExecutor.isPresent()) {
+                return Response.failure(ServiceState.PLUGIN_NOT_FOUND);
+            }
+
+            SourceEntity toSource = toSourceOptional.get();
+            IConfigure toConfigure = PluginCommon.loadConfigure(toSource.getProtocol(), toSource.getType(), toSource.getType(), environment);
+            Optional<IConfigureExecutor> toConfigureExecutor = toConfigure.getPipelines()
+                    .stream()
+                    .filter(v -> v.getExecutor().equals(configure.getExecutor()))
+                    .findFirst();
+            if (!toConfigureExecutor.isPresent()) {
+                return Response.failure(ServiceState.PLUGIN_NOT_FOUND);
+            }
+
+            Optional<Executor> executorOptional = injector.getInstance(Key.get(new TypeLiteral<Set<Executor>>() {}))
+                    .stream()
+                    .filter(executor -> executor.name().equals(configure.getExecutor()))
+                    .findFirst();
+
+            // FROM source
+            Properties fromProperties = new Properties();
+            Properties fromBeanProperties = BeanToPropertiesCommon.convertBeanToProperties(fromSource);
+            fromConfigureExecutor.get().getFields().forEach(pipelineField -> fromProperties.put(pipelineField.getField(), fromBeanProperties.get(pipelineField.getField())));
+            PipelineField fromField = PipelineField.builder()
+                    .type(fromSource.getType())
+                    .configure(fromProperties)
+                    .build();
+
+            // TO source
+            Properties toProperties = new Properties();
+            Properties toBeanProperties = BeanToPropertiesCommon.convertBeanToProperties(toSource);
+            toConfigureExecutor.get().getFields().forEach(pipelineField -> toProperties.put(pipelineField.getField(), toBeanProperties.get(pipelineField.getField())));
+            PipelineField toField = PipelineField.builder()
+                    .type(toSource.getType())
+                    .configure(toProperties)
+                    .build();
+
+            Pipeline pipeline = Pipeline.builder()
+                    .work(System.getenv("user.dir") + "/" + UUID.randomUUID().toString())
+                    .home(environment.getProperty("datacap.executor.seatunnel.home"))
+                    .from(fromField)
+                    .to(toField)
+                    .build();
+            executorOptional.get().start(pipeline);
+            return Response.success(null);
+        }
+        return Response.failure(ServiceState.SOURCE_NOT_FOUND);
+    }
+}