diff --git a/.github/workflows/e2e.yml b/.github/workflows/e2e.yml
index b32b860b67..8a887d93a6 100644
--- a/.github/workflows/e2e.yml
+++ b/.github/workflows/e2e.yml
@@ -110,6 +110,12 @@ jobs:
     strategy:
       matrix:
         case:
+          - name: Flink116OnRemoteClusterDeployTest
+            class: org.apache.streampark.e2e.cases.Flink118OnRemoteClusterDeployTest
+          - name: Flink116OnRemoteClusterDeployTest
+            class: org.apache.streampark.e2e.cases.Flink117OnRemoteClusterDeployTest
+          - name: Flink116OnRemoteClusterDeployTest
+            class: org.apache.streampark.e2e.cases.Flink116OnRemoteClusterDeployTest
           - name: EnvironmentTest
             class: org.apache.streampark.e2e.cases.EnvironmentTest
           - name: AlarmTest
diff --git a/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink116OnRemoteClusterDeployTest.java b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink116OnRemoteClusterDeployTest.java
new file mode 100644
index 0000000000..1821b3c5ab
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink116OnRemoteClusterDeployTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.e2e.cases;
+
+import org.apache.streampark.e2e.core.StreamPark;
+import org.apache.streampark.e2e.pages.LoginPage;
+import org.apache.streampark.e2e.pages.common.Constants;
+import org.apache.streampark.e2e.pages.flink.ApacheFlinkPage;
+import org.apache.streampark.e2e.pages.flink.FlinkHomePage;
+import org.apache.streampark.e2e.pages.flink.clusters.ClusterDetailForm;
+import org.apache.streampark.e2e.pages.flink.clusters.FlinkClustersPage;
+import org.apache.streampark.e2e.pages.flink.clusters.RemoteForm;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.openqa.selenium.WebElement;
+import org.openqa.selenium.remote.RemoteWebDriver;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@StreamPark(composeFiles = "docker/flink-1.16-on-remote/docker-compose.yaml")
+public class Flink116OnRemoteClusterDeployTest {
+
+    private static RemoteWebDriver browser;
+
+    private static final String userName = "admin";
+
+    private static final String password = "streampark";
+
+    private static final String teamName = "default";
+
+    private static final String flinkName = "flink-1.16.3";
+
+    private static final String flinkHome = "/opt/flink/";
+
+    private static final String flinkDescription = "description test";
+
+    private static final String flinkClusterName = "flink_1.16.3_cluster_e2e";
+
+    private static final String flinkJobManagerUrl = "http://jobmanager:8081";
+
+    private static final ClusterDetailForm.ExecutionMode executionMode = ClusterDetailForm.ExecutionMode.REMOTE;
+
+    @BeforeAll
+    public static void setUp() {
+        FlinkHomePage flinkHomePage = new LoginPage(browser)
+            .login(userName, password, teamName)
+            .goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkHomePage.class);
+
+        flinkHomePage.createFlinkHome(flinkName, flinkHome, flinkDescription);
+
+        flinkHomePage.goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkClustersPage.class);
+    }
+
+    @Test
+    @Order(10)
+    public void testCreateFlinkCluster() {
+        FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.createFlinkCluster()
+            .<RemoteForm>addCluster(executionMode)
+            .jobManagerURL(flinkJobManagerUrl)
+            .clusterName(flinkClusterName)
+            .flinkVersion(flinkName)
+            .submit();
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> assertThat(flinkClustersPage.flinkClusterList())
+                    .as("Flink clusters list should contain newly-created application")
+                    .extracting(WebElement::getText)
+                    .anyMatch(it -> it.contains(flinkClusterName)));
+    }
+
+    @Test
+    @Order(50)
+    public void testDeleteFlinkCluster() {
+        final FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.deleteFlinkCluster(flinkClusterName);
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> {
+                    browser.navigate().refresh();
+                    Thread.sleep(Constants.DEFAULT_SLEEP_MILLISECONDS);
+                    assertThat(flinkClustersPage.flinkClusterList())
+                        .noneMatch(it -> it.getText().contains(flinkClusterName));
+                });
+    }
+}
diff --git a/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink117OnRemoteClusterDeployTest.java b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink117OnRemoteClusterDeployTest.java
new file mode 100644
index 0000000000..58f93e34a2
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink117OnRemoteClusterDeployTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.e2e.cases;
+
+import org.apache.streampark.e2e.core.StreamPark;
+import org.apache.streampark.e2e.pages.LoginPage;
+import org.apache.streampark.e2e.pages.common.Constants;
+import org.apache.streampark.e2e.pages.flink.ApacheFlinkPage;
+import org.apache.streampark.e2e.pages.flink.FlinkHomePage;
+import org.apache.streampark.e2e.pages.flink.clusters.ClusterDetailForm;
+import org.apache.streampark.e2e.pages.flink.clusters.FlinkClustersPage;
+import org.apache.streampark.e2e.pages.flink.clusters.RemoteForm;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.openqa.selenium.WebElement;
+import org.openqa.selenium.remote.RemoteWebDriver;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@StreamPark(composeFiles = "docker/flink-1.17-on-remote/docker-compose.yaml")
+public class Flink117OnRemoteClusterDeployTest {
+
+    private static RemoteWebDriver browser;
+
+    private static final String userName = "admin";
+
+    private static final String password = "streampark";
+
+    private static final String teamName = "default";
+
+    private static final String flinkName = "flink-1.17.2";
+
+    private static final String flinkHome = "/opt/flink/";
+
+    private static final String flinkDescription = "description test";
+
+    private static final String flinkClusterName = "flink_1.17.2_cluster_e2e";
+
+    private static final String flinkJobManagerUrl = "http://jobmanager:8081";
+
+    private static final ClusterDetailForm.ExecutionMode executionMode = ClusterDetailForm.ExecutionMode.REMOTE;
+
+    @BeforeAll
+    public static void setUp() {
+        FlinkHomePage flinkHomePage = new LoginPage(browser)
+            .login(userName, password, teamName)
+            .goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkHomePage.class);
+
+        flinkHomePage.createFlinkHome(flinkName, flinkHome, flinkDescription);
+
+        flinkHomePage.goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkClustersPage.class);
+    }
+
+    @Test
+    @Order(10)
+    public void testCreateFlinkCluster() {
+        FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.createFlinkCluster()
+            .<RemoteForm>addCluster(executionMode)
+            .jobManagerURL(flinkJobManagerUrl)
+            .clusterName(flinkClusterName)
+            .flinkVersion(flinkName)
+            .submit();
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> assertThat(flinkClustersPage.flinkClusterList())
+                    .as("Flink clusters list should contain newly-created application")
+                    .extracting(WebElement::getText)
+                    .anyMatch(it -> it.contains(flinkClusterName)));
+    }
+
+    @Test
+    @Order(50)
+    public void testDeleteFlinkCluster() {
+        final FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.deleteFlinkCluster(flinkClusterName);
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> {
+                    browser.navigate().refresh();
+                    Thread.sleep(Constants.DEFAULT_SLEEP_MILLISECONDS);
+                    assertThat(flinkClustersPage.flinkClusterList())
+                        .noneMatch(it -> it.getText().contains(flinkClusterName));
+                });
+    }
+}
diff --git a/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink118OnRemoteClusterDeployTest.java b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink118OnRemoteClusterDeployTest.java
new file mode 100644
index 0000000000..a4aa8a231f
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/cases/Flink118OnRemoteClusterDeployTest.java
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.e2e.cases;
+
+import org.apache.streampark.e2e.core.StreamPark;
+import org.apache.streampark.e2e.pages.LoginPage;
+import org.apache.streampark.e2e.pages.common.Constants;
+import org.apache.streampark.e2e.pages.flink.ApacheFlinkPage;
+import org.apache.streampark.e2e.pages.flink.FlinkHomePage;
+import org.apache.streampark.e2e.pages.flink.clusters.ClusterDetailForm;
+import org.apache.streampark.e2e.pages.flink.clusters.FlinkClustersPage;
+import org.apache.streampark.e2e.pages.flink.clusters.RemoteForm;
+
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.Order;
+import org.junit.jupiter.api.Test;
+import org.openqa.selenium.WebElement;
+import org.openqa.selenium.remote.RemoteWebDriver;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+@StreamPark(composeFiles = "docker/flink-1.18-on-remote/docker-compose.yaml")
+public class Flink118OnRemoteClusterDeployTest {
+
+    private static RemoteWebDriver browser;
+
+    private static final String userName = "admin";
+
+    private static final String password = "streampark";
+
+    private static final String teamName = "default";
+
+    private static final String flinkName = "flink-1.18.1";
+
+    private static final String flinkHome = "/opt/flink/";
+
+    private static final String flinkDescription = "description test";
+
+    private static final String flinkClusterName = "flink_1.18.1_cluster_e2e";
+
+    private static final String flinkJobManagerUrl = "http://jobmanager:8081";
+
+    private static final ClusterDetailForm.ExecutionMode executionMode = ClusterDetailForm.ExecutionMode.REMOTE;
+
+    @BeforeAll
+    public static void setUp() {
+        FlinkHomePage flinkHomePage = new LoginPage(browser)
+            .login(userName, password, teamName)
+            .goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkHomePage.class);
+
+        flinkHomePage.createFlinkHome(flinkName, flinkHome, flinkDescription);
+
+        flinkHomePage.goToNav(ApacheFlinkPage.class)
+            .goToTab(FlinkClustersPage.class);
+    }
+
+    @Test
+    @Order(10)
+    public void testCreateFlinkCluster() {
+        FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.createFlinkCluster()
+            .<RemoteForm>addCluster(executionMode)
+            .jobManagerURL(flinkJobManagerUrl)
+            .clusterName(flinkClusterName)
+            .flinkVersion(flinkName)
+            .submit();
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> assertThat(flinkClustersPage.flinkClusterList())
+                    .as("Flink clusters list should contain newly-created application")
+                    .extracting(WebElement::getText)
+                    .anyMatch(it -> it.contains(flinkClusterName)));
+    }
+
+    @Test
+    @Order(50)
+    public void testDeleteFlinkCluster() {
+        final FlinkClustersPage flinkClustersPage = new FlinkClustersPage(browser);
+
+        flinkClustersPage.deleteFlinkCluster(flinkClusterName);
+
+        Awaitility.await()
+            .untilAsserted(
+                () -> {
+                    browser.navigate().refresh();
+                    Thread.sleep(Constants.DEFAULT_SLEEP_MILLISECONDS);
+                    assertThat(flinkClustersPage.flinkClusterList())
+                        .noneMatch(it -> it.getText().contains(flinkClusterName));
+                });
+    }
+}
diff --git a/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/pages/flink/clusters/ClusterDetailForm.java b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/pages/flink/clusters/ClusterDetailForm.java
index 15ad80de68..deb28d55f2 100644
--- a/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/pages/flink/clusters/ClusterDetailForm.java
+++ b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/pages/flink/clusters/ClusterDetailForm.java
@@ -50,6 +50,14 @@ public ClusterDetailForm(WebDriver driver) {
     public <T> T addCluster(ExecutionMode executionMode) {
         buttonExecutionModeDropdown.click();
         switch (executionMode) {
+            case REMOTE:
+                selectExecutionMode.stream()
+                    .filter(e -> e.getText().equalsIgnoreCase(ExecutionMode.REMOTE.desc()))
+                    .findFirst()
+                    .orElseThrow(() -> new IllegalArgumentException(
+                        String.format("Execution Mode not found: %s", executionMode.desc())))
+                    .click();
+                return (T) new RemoteForm(this);
             case YARN_SESSION:
                 selectExecutionMode.stream()
                     .filter(e -> e.getText().equalsIgnoreCase(ExecutionMode.YARN_SESSION.desc()))
diff --git a/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/pages/flink/clusters/RemoteForm.java b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/pages/flink/clusters/RemoteForm.java
new file mode 100644
index 0000000000..31f9bce727
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/java/org/apache/streampark/e2e/pages/flink/clusters/RemoteForm.java
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.streampark.e2e.pages.flink.clusters;
+
+import org.apache.streampark.e2e.pages.common.CommonFactory;
+
+import lombok.Getter;
+import org.openqa.selenium.WebDriver;
+import org.openqa.selenium.WebElement;
+import org.openqa.selenium.support.FindBy;
+
+@Getter
+public class RemoteForm extends CommonForm {
+
+    private WebDriver driver;
+
+    @FindBy(id = "form_item_address")
+    private WebElement inputJobManagerURL;
+
+    public RemoteForm(ClusterDetailForm clusterDetailForm) {
+        super(clusterDetailForm);
+
+        this.driver = clusterDetailForm.driver();
+    }
+
+    public RemoteForm jobManagerURL(String address) {
+        CommonFactory.WebElementDeleteAndInput(this.inputJobManagerURL, address);
+        return this;
+    }
+}
diff --git a/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.16-on-remote/Dockerfile b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.16-on-remote/Dockerfile
new file mode 100644
index 0000000000..b1e093fd41
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.16-on-remote/Dockerfile
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+FROM apache/streampark:ci
diff --git a/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.16-on-remote/docker-compose.yaml b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.16-on-remote/docker-compose.yaml
new file mode 100644
index 0000000000..be496ff7c0
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.16-on-remote/docker-compose.yaml
@@ -0,0 +1,90 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+services:
+  jobmanager:
+    image: flink:1.16.3
+    command: jobmanager
+    ports:
+      - "8081:8081"
+    environment:
+      - |
+        FLINK_PROPERTIES=
+        jobmanager.rpc.address: jobmanager
+    networks:
+      - e2e
+    volumes:
+      - flink_data:/opt/flink
+      - /var/run/docker.sock:/var/run/docker.sock
+    healthcheck:
+      test: [ "CMD", "curl", "http://localhost:8081" ]
+      interval: 5s
+      timeout: 5s
+      retries: 120
+
+  taskmanager:
+    image: flink:1.16.3
+    depends_on:
+      - jobmanager
+    command: taskmanager
+    scale: 1
+    environment:
+      - |
+        FLINK_PROPERTIES=
+        jobmanager.rpc.address: jobmanager
+        taskmanager.numberOfTaskSlots: 2
+    networks:
+      - e2e
+    volumes:
+      - flink_data:/opt/flink
+      - /var/run/docker.sock:/var/run/docker.sock
+    healthcheck:
+      test: [ "CMD", "curl", "http://localhost:8081" ]
+      interval: 5s
+      timeout: 5s
+      retries: 120
+
+  streampark:
+    image: apache/streampark:ci
+    command: bash bin/streampark.sh start_docker
+    build:
+      context: ./
+      dockerfile: ./Dockerfile
+    ports:
+      - 10000:10000
+      - 10030:10030
+    environment:
+      - SPRING_PROFILES_ACTIVE=h2
+      - TZ=Asia/Shanghai
+      - FLINK_JOBMANAGER_URL=http://jobmanager:8081
+    privileged: true
+    restart: unless-stopped
+    networks:
+      - e2e
+    volumes:
+      - flink_data:/opt/flink
+      - ${HOME}/streampark_build_logs:/tmp/streampark/logs/build_logs/
+      - /var/run/docker.sock:/var/run/docker.sock
+    healthcheck:
+      test: [ "CMD", "curl", "http://localhost:10000" ]
+      interval: 5s
+      timeout: 5s
+      retries: 120
+networks:
+  e2e:
+volumes:
+  flink_data:
diff --git a/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.17-on-remote/Dockerfile b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.17-on-remote/Dockerfile
new file mode 100644
index 0000000000..b1e093fd41
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.17-on-remote/Dockerfile
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+FROM apache/streampark:ci
diff --git a/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.17-on-remote/docker-compose.yaml b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.17-on-remote/docker-compose.yaml
new file mode 100644
index 0000000000..724172e32d
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.17-on-remote/docker-compose.yaml
@@ -0,0 +1,90 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+services:
+  jobmanager:
+    image: flink:1.17.2
+    command: jobmanager
+    ports:
+      - "8081:8081"
+    environment:
+      - |
+        FLINK_PROPERTIES=
+        jobmanager.rpc.address: jobmanager
+    networks:
+      - e2e
+    volumes:
+      - flink_data:/opt/flink
+      - /var/run/docker.sock:/var/run/docker.sock
+    healthcheck:
+      test: [ "CMD", "curl", "http://localhost:8081" ]
+      interval: 5s
+      timeout: 5s
+      retries: 120
+
+  taskmanager:
+    image: flink:1.17.2
+    depends_on:
+      - jobmanager
+    command: taskmanager
+    scale: 1
+    environment:
+      - |
+        FLINK_PROPERTIES=
+        jobmanager.rpc.address: jobmanager
+        taskmanager.numberOfTaskSlots: 2
+    networks:
+      - e2e
+    volumes:
+      - flink_data:/opt/flink
+      - /var/run/docker.sock:/var/run/docker.sock
+    healthcheck:
+      test: [ "CMD", "curl", "http://localhost:8081" ]
+      interval: 5s
+      timeout: 5s
+      retries: 120
+
+  streampark:
+    image: apache/streampark:ci
+    command: bash bin/streampark.sh start_docker
+    build:
+      context: ./
+      dockerfile: ./Dockerfile
+    ports:
+      - 10000:10000
+      - 10030:10030
+    environment:
+      - SPRING_PROFILES_ACTIVE=h2
+      - TZ=Asia/Shanghai
+      - FLINK_JOBMANAGER_URL=http://jobmanager:8081
+    privileged: true
+    restart: unless-stopped
+    networks:
+      - e2e
+    volumes:
+      - flink_data:/opt/flink
+      - ${HOME}/streampark_build_logs:/tmp/streampark/logs/build_logs/
+      - /var/run/docker.sock:/var/run/docker.sock
+    healthcheck:
+      test: [ "CMD", "curl", "http://localhost:10000" ]
+      interval: 5s
+      timeout: 5s
+      retries: 120
+networks:
+  e2e:
+volumes:
+  flink_data:
diff --git a/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.18-on-remote/Dockerfile b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.18-on-remote/Dockerfile
new file mode 100644
index 0000000000..b1e093fd41
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.18-on-remote/Dockerfile
@@ -0,0 +1,18 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+FROM apache/streampark:ci
diff --git a/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.18-on-remote/docker-compose.yaml b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.18-on-remote/docker-compose.yaml
new file mode 100644
index 0000000000..f44d1af386
--- /dev/null
+++ b/streampark-e2e/streampark-e2e-case/src/test/resources/docker/flink-1.18-on-remote/docker-compose.yaml
@@ -0,0 +1,90 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+services:
+  jobmanager:
+    image: flink:1.18.1
+    command: jobmanager
+    ports:
+      - "8081:8081"
+    environment:
+      - |
+        FLINK_PROPERTIES=
+        jobmanager.rpc.address: jobmanager
+    networks:
+      - e2e
+    volumes:
+      - flink_data:/opt/flink
+      - /var/run/docker.sock:/var/run/docker.sock
+    healthcheck:
+      test: [ "CMD", "curl", "http://localhost:8081" ]
+      interval: 5s
+      timeout: 5s
+      retries: 120
+
+  taskmanager:
+    image: flink:1.18.1
+    depends_on:
+      - jobmanager
+    command: taskmanager
+    scale: 1
+    environment:
+      - |
+        FLINK_PROPERTIES=
+        jobmanager.rpc.address: jobmanager
+        taskmanager.numberOfTaskSlots: 2
+    networks:
+      - e2e
+    volumes:
+      - flink_data:/opt/flink
+      - /var/run/docker.sock:/var/run/docker.sock
+    healthcheck:
+      test: [ "CMD", "curl", "http://localhost:8081" ]
+      interval: 5s
+      timeout: 5s
+      retries: 120
+
+  streampark:
+    image: apache/streampark:ci
+    command: bash bin/streampark.sh start_docker
+    build:
+      context: ./
+      dockerfile: ./Dockerfile
+    ports:
+      - 10000:10000
+      - 10030:10030
+    environment:
+      - SPRING_PROFILES_ACTIVE=h2
+      - TZ=Asia/Shanghai
+      - FLINK_JOBMANAGER_URL=http://jobmanager:8081
+    privileged: true
+    restart: unless-stopped
+    networks:
+      - e2e
+    volumes:
+      - flink_data:/opt/flink
+      - ${HOME}/streampark_build_logs:/tmp/streampark/logs/build_logs/
+      - /var/run/docker.sock:/var/run/docker.sock
+    healthcheck:
+      test: [ "CMD", "curl", "http://localhost:10000" ]
+      interval: 5s
+      timeout: 5s
+      retries: 120
+networks:
+  e2e:
+volumes:
+  flink_data: