This is an automated email from the ASF dual-hosted git repository.
wanghailin pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 458d3632b1 [Feature] Support Zeta engine on Kubernetes (#5594)
458d3632b1 is described below
commit 458d3632b10dc8310a325804116538a2422f204d
Author: xiami <[email protected]>
AuthorDate: Fri Nov 3 10:22:04 2023 +0800
[Feature] Support Zeta engine on Kubernetes (#5594)
---
.github/workflows/backend.yml | 33 +-
docs/en/start-v2/kubernetes/kubernetes.mdx | 487 ++++++++++++++++++++-
seatunnel-e2e/seatunnel-engine-e2e/pom.xml | 1 +
.../{ => seatunnel-engine-k8s-e2e}/pom.xml | 55 +--
.../seatunnel/engine/e2e/k8s/KubernetesIT.java | 231 ++++++++++
.../resources/custom_config/hazelcast-client.yaml | 23 +
.../test/resources/custom_config/hazelcast.yaml | 45 ++
.../custom_config/plugin-mapping.properties | 27 ++
.../src/test/resources/seatunnel-service.yaml | 29 ++
.../src/test/resources/seatunnel-statefulset.yaml | 50 +++
.../src/test/resources/seatunnel_dockerfile | 30 ++
11 files changed, 970 insertions(+), 41 deletions(-)
diff --git a/.github/workflows/backend.yml b/.github/workflows/backend.yml
index 2b13cefc5c..48b659974c 100644
--- a/.github/workflows/backend.yml
+++ b/.github/workflows/backend.yml
@@ -543,7 +543,38 @@ jobs:
./mvnw -T 1 -B verify -DskipUT=true -DskipIT=false
-D"license.skipAddThirdParty"=true --no-snapshot-updates -pl
:connector-seatunnel-e2e-base -am -Pci
env:
MAVEN_OPTS: -Xmx4096m
-
+ engine-k8s-it:
+ needs: [ changes, sanity-check ]
+ if: needs.changes.outputs.api == 'true'
+ runs-on: ${{ matrix.os }}
+ strategy:
+ matrix:
+ java: [ '8', '11' ]
+ os: [ 'ubuntu-latest' ]
+ timeout-minutes: 45
+ steps:
+ - name: install k8s
+ run: |
+ curl -sfL https://get.k3s.io | K3S_KUBECONFIG_MODE=777 sh -s -
--docker
+ cat /etc/rancher/k3s/k3s.yaml
+ mkdir -p ~/.kube
+ cp /etc/rancher/k3s/k3s.yaml ~/.kube/config
+ env:
+ KUBECONFIG: /etc/rancher/k3s/k3s.yaml
+ - uses: actions/checkout@v2
+ - name: Set up JDK ${{ matrix.java }}
+ uses: actions/setup-java@v3
+ with:
+ java-version: ${{ matrix.java }}
+ distribution: 'temurin'
+ cache: 'maven'
+ - name: run seatunnel zeta on k8s test
+ if: needs.changes.outputs.api == 'true'
+ run: |
+ ./mvnw -T 1 -B verify -DskipUT=true -DskipIT=false
-D"license.skipAddThirdParty"=true --no-snapshot-updates -pl
:seatunnel-engine-k8s-e2e -am -Pci
+ env:
+ MAVEN_OPTS: -Xmx4096m
+ KUBECONFIG: /etc/rancher/k3s/k3s.yaml
transform-v2-it-part-1:
needs: [ changes, sanity-check ]
if: needs.changes.outputs.api == 'true'
diff --git a/docs/en/start-v2/kubernetes/kubernetes.mdx
b/docs/en/start-v2/kubernetes/kubernetes.mdx
index 6ba479aa4f..641793cdac 100644
--- a/docs/en/start-v2/kubernetes/kubernetes.mdx
+++ b/docs/en/start-v2/kubernetes/kubernetes.mdx
@@ -33,9 +33,11 @@ To run the image with SeaTunnel, first create a `Dockerfile`:
<Tabs
groupId="engine-type"
- defaultValue="flink"
+ defaultValue="Zeta (local-mode)"
values={[
{label: 'Flink', value: 'flink'},
+ {label: 'Zeta (local-mode)', value: 'Zeta (local-mode)'},
+ {label: 'Zeta (cluster-mode)', value: 'Zeta (cluster-mode)'},
]}>
<TabItem value="flink">
@@ -63,16 +65,75 @@ Load image to minikube via:
minikube image load seatunnel:2.3.0-flink-1.13
```
+</TabItem>
+
+<TabItem value="Zeta (local-mode)">
+
+```Dockerfile
+FROM openjdk:8
+
+ENV SEATUNNEL_VERSION="2.3.3"
+ENV SEATUNNEL_HOME="/opt/seatunnel"
+
+RUN wget
https://dlcdn.apache.org/seatunnel/${SEATUNNEL_VERSION}/apache-seatunnel-${SEATUNNEL_VERSION}-bin.tar.gz
+RUN tar -xzvf apache-seatunnel-${SEATUNNEL_VERSION}-bin.tar.gz
+RUN mv apache-seatunnel-${SEATUNNEL_VERSION} ${SEATUNNEL_HOME}
+
+RUN cd ${SEATUNNEL_HOME}||sh bin/install-plugin.sh ${SEATUNNEL_VERSION}
+```
+
+Then run the following commands to build the image:
+```bash
+docker build -t seatunnel:2.3.3 -f Dockerfile .
+```
+Image `seatunnel:2.3.3` need to be present in the host (minikube) so that the
deployment can take place.
+
+Load image to minikube via:
+```bash
+minikube image load seatunnel:2.3.3
+```
+
+</TabItem>
+
+<TabItem value="Zeta (cluster-mode)">
+
+```Dockerfile
+FROM openjdk:8
+
+ENV SEATUNNEL_VERSION="2.3.3"
+ENV SEATUNNEL_HOME="/opt/seatunnel"
+
+RUN wget
https://dlcdn.apache.org/seatunnel/${SEATUNNEL_VERSION}/apache-seatunnel-${SEATUNNEL_VERSION}-bin.tar.gz
+RUN tar -xzvf apache-seatunnel-${SEATUNNEL_VERSION}-bin.tar.gz
+RUN mv apache-seatunnel-${SEATUNNEL_VERSION} ${SEATUNNEL_HOME}
+RUN mkdir -p $SEATUNNEL_HOME/logs
+RUN cd ${SEATUNNEL_HOME}||sh bin/install-plugin.sh ${SEATUNNEL_VERSION}
+```
+
+Then run the following commands to build the image:
+```bash
+docker build -t seatunnel:2.3.3 -f Dockerfile .
+```
+Image `seatunnel:2.3.3` need to be present in the host (minikube) so that the
deployment can take place.
+
+Load image to minikube via:
+```bash
+minikube image load seatunnel:2.3.3
+```
+
</TabItem>
</Tabs>
+
### Deploying the operator
<Tabs
groupId="engine-type"
- defaultValue="flink"
+ defaultValue="Zeta (local-mode)"
values={[
{label: 'Flink', value: 'flink'},
+ {label: 'Zeta (local-mode)', value: 'Zeta (local-mode)'},
+ {label: 'Zeta (cluster-mode)', value: 'Zeta (cluster-mode)'},
]}>
<TabItem value="flink">
@@ -105,6 +166,15 @@ flink-kubernetes-operator-5f466b8549-mgchb 1/1
Running 3 (23h
```
</TabItem>
+
+
+<TabItem value="Zeta (local-mode)">
+none
+</TabItem>
+
+<TabItem value="Zeta (cluster-mode)">
+none
+</TabItem>
</Tabs>
## Run SeaTunnel Application
@@ -113,9 +183,11 @@ flink-kubernetes-operator-5f466b8549-mgchb 1/1
Running 3 (23h
<Tabs
groupId="engine-type"
- defaultValue="flink"
+ defaultValue="Zeta (local-mode)"
values={[
{label: 'Flink', value: 'flink'},
+ {label: 'Zeta (local-mode)', value: 'Zeta (local-mode)'},
+ {label: 'Zeta (cluster-mode)', value: 'Zeta (cluster-mode)'},
]}>
<TabItem value="flink">
@@ -216,15 +288,346 @@ kubectl apply -f seatunnel-flink.yaml
```
</TabItem>
+
+<TabItem value="Zeta (local-mode)">
+
+In this guide we are going to use
[seatunnel.streaming.conf](https://github.com/apache/seatunnel/blob/2.3.0-release/config/v2.streaming.conf.template):
+
+```conf
+env {
+ execution.parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 2000
+}
+
+source {
+ FakeSource {
+ parallelism = 2
+ result_table_name = "fake"
+ row.num = 16
+ schema = {
+ fields {
+ name = "string"
+ age = "int"
+ }
+ }
+ }
+}
+
+sink {
+ Console {
+ }
+}
+```
+
+Generate a configmap named seatunnel-config in Kubernetes for the
seatunnel.streaming.conf so that we can mount the config content in pod.
+```bash
+kubectl create cm seatunnel-config \
+--from-file=seatunnel.streaming.conf=seatunnel.streaming.conf
+```
+- Create `seatunnel.yaml`:
+```yaml
+apiVersion: v1
+kind: Pod
+metadata:
+ name: seatunnel
+spec:
+ containers:
+ - name: seatunnel
+ image: seatunnel:2.3.3
+ command: ["/bin/sh","-c","/opt/seatunnel/bin/seatunnel.sh --config
/data/seatunnel.streaming.conf -e local"]
+ resources:
+ limits:
+ cpu: "1"
+ memory: 4G
+ requests:
+ cpu: "1"
+ memory: 2G
+ volumeMounts:
+ - name: seatunnel-config
+ mountPath: /data/seatunnel.streaming.conf
+ subPath: seatunnel.streaming.conf
+ volumes:
+ - name: seatunnel-config
+ configMap:
+ name: seatunnel-config
+ items:
+ - key: seatunnel.streaming.conf
+ path: seatunnel.streaming.conf
+```
+
+- Run the example application:
+```bash
+kubectl apply -f seatunnel.yaml
+```
+
+</TabItem>
+
+
+<TabItem value="Zeta (cluster-mode)">
+
+In this guide we are going to use
[seatunnel.streaming.conf](https://github.com/apache/seatunnel/blob/2.3.0-release/config/v2.streaming.conf.template):
+
+```conf
+env {
+ execution.parallelism = 2
+ job.mode = "STREAMING"
+ checkpoint.interval = 2000
+}
+
+source {
+ FakeSource {
+ parallelism = 2
+ result_table_name = "fake"
+ row.num = 16
+ schema = {
+ fields {
+ name = "string"
+ age = "int"
+ }
+ }
+ }
+}
+
+sink {
+ Console {
+ }
+}
+```
+
+Generate a configmap named seatunnel-config in Kubernetes for the
seatunnel.streaming.conf so that we can mount the config content in pod.
+```bash
+kubectl create cm seatunnel-config \
+--from-file=seatunnel.streaming.conf=seatunnel.streaming.conf
+```
+
+Then, we use the following command to load some configuration files used by
the seatunnel cluster into the configmap
+
+Create the yaml file locally as follows
+
+- Create `hazelcast-client.yaml`:
+
+```yaml
+
+hazelcast-client:
+ cluster-name: seatunnel
+ properties:
+ hazelcast.logging.type: log4j2
+ network:
+ cluster-members:
+ - localhost:5801
+
+```
+- Create `hazelcast.yaml`:
+
+```yaml
+
+hazelcast:
+ cluster-name: seatunnel
+ network:
+ rest-api:
+ enabled: true
+ endpoint-groups:
+ CLUSTER_WRITE:
+ enabled: true
+ DATA:
+ enabled: true
+ join:
+ tcp-ip:
+ enabled: true
+ member-list:
+ - localhost
+ port:
+ auto-increment: false
+ port: 5801
+ properties:
+ hazelcast.invocation.max.retry.count: 20
+ hazelcast.tcp.join.port.try.count: 30
+ hazelcast.logging.type: log4j2
+ hazelcast.operation.generic.thread.count: 50
+
+```
+- Create `seatunnel.yaml`:
+
+```yaml
+seatunnel:
+ engine:
+ history-job-expire-minutes: 1440
+ backup-count: 1
+ queue-type: blockingqueue
+ print-execution-info-interval: 60
+ print-job-metrics-info-interval: 60
+ slot-service:
+ dynamic-slot: true
+ checkpoint:
+ interval: 10000
+ timeout: 60000
+ storage:
+ type: hdfs
+ max-retained: 3
+ plugin-config:
+ namespace: /tmp/seatunnel/checkpoint_snapshot
+ storage.type: hdfs
+ fs.defaultFS: file:///tmp/ # Ensure that the directory has written
permission
+```
+
+Create congfigmaps for the configuration file using the following command
+
+```bash
+kubectl create configmap hazelcast-client --from-file=hazelcast-client.yaml
+kubectl create configmap hazelcast --from-file=hazelcast.yaml
+kubectl create configmap seatunnelmap --from-file=seatunnel.yaml
+
+```
+
+Deploy Reloader to achieve hot deployment
+We use the Reloader here to automatically restart the pod when the
configuration file or other modifications are made. You can also directly give
the value of the configuration file and do not use the Reloader
+
+- [Reloader](https://github.com/stakater/Reloader/)
+
+```bash
+wget
https://raw.githubusercontent.com/stakater/Reloader/master/deployments/kubernetes/reloader.yaml
+kubectl apply -f reloader.yaml
+
+```
+
+- Create `seatunnel-cluster.yml`:
+```yaml
+apiVersion: v1
+kind: Service
+metadata:
+ name: seatunnel
+spec:
+ selector:
+ app: seatunnel
+ ports:
+ - port: 5801
+ name: seatunnel
+ clusterIP: None
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+ name: seatunnel
+ annotations:
+ configmap.reloader.stakater.com/reload:
"hazelcast,hazelcast-client,seatunnelmap"
+spec:
+ serviceName: "seatunnel"
+ replicas: 3 # modify replicas according to your case
+ selector:
+ matchLabels:
+ app: seatunnel
+ template:
+ metadata:
+ labels:
+ app: seatunnel
+ spec:
+ containers:
+ - name: seatunnel
+ image: seatunnel:2.3.3
+ imagePullPolicy: IfNotPresent
+ ports:
+ - containerPort: 5801
+ name: client
+ command: ["/bin/sh","-c","/opt/seatunnel/bin/seatunnel-cluster.sh
-DJvmOption=-Xms2G -Xmx2G"]
+ resources:
+ limits:
+ cpu: "1"
+ memory: 4G
+ requests:
+ cpu: "1"
+ memory: 2G
+ volumeMounts:
+ - mountPath: "/opt/seatunnel/config/hazelcast.yaml"
+ name: hazelcast
+ subPath: hazelcast.yaml
+ - mountPath: "/opt/seatunnel/config/hazelcast-client.yaml"
+ name: hazelcast-client
+ subPath: hazelcast-client.yaml
+ - mountPath: "/opt/seatunnel/config/seatunnel.yaml"
+ name: seatunnelmap
+ subPath: seatunnel.yaml
+ - mountPath: /data/seatunnel.streaming.conf
+ name: seatunnel-config
+ subPath: seatunnel.streaming.conf
+ volumes:
+ - name: hazelcast
+ configMap:
+ name: hazelcast
+ - name: hazelcast-client
+ configMap:
+ name: hazelcast-client
+ - name: seatunnelmap
+ configMap:
+ name: seatunnelmap
+ - name: seatunnel-config
+ configMap:
+ name: seatunnel-config
+ items:
+ - key: seatunnel.streaming.conf
+ path: seatunnel.streaming.conf
+```
+
+- Starting a cluster:
+```bash
+kubectl apply -f seatunnel-cluster.yml
+```
+Then modify the seatunnel configuration in pod using the following command
+
+```bash
+kubectl edit cm hazelcast
+```
+Change the member-list option to your cluster address
+
+This uses the headless service access mode
+
+The format for accessing between general pods is
[pod-name].[service-name].[namespace].svc.cluster.local
+
+for example:
+```bash
+- seatunnel-0.seatunnel.default.svc.cluster.local
+- seatunnel-1.seatunnel.default.svc.cluster.local
+- seatunnel-2.seatunnel.default.svc.cluster.local
+```
+```bash
+kubectl edit cm hazelcast-client
+```
+Change the cluster-members option to your cluster address
+
+for example:
+```bash
+- seatunnel-0.seatunnel.default.svc.cluster.local:5801
+- seatunnel-1.seatunnel.default.svc.cluster.local:5801
+- seatunnel-2.seatunnel.default.svc.cluster.local:5801
+```
+Later, you will see that the pod automatically restarts and updates the
seatunnel configuration
+
+```bash
+kubectl edit cm hazelcast-client
+```
+After we wait for all pod updates to be completed, we can use the following
command to check if the configuration inside the pod has been updated
+
+```bash
+kubectl exec -it seatunnel-0 -- cat
/opt/seatunnel/config/hazelcast-client.yaml
+```
+Afterwards, we can submit tasks to any pod
+
+```bash
+kubectl exec -it seatunnel-0 -- /opt/seatunnel/bin/seatunnel.sh --config
/data/seatunnel.streaming.conf
+```
+</TabItem>
+
</Tabs>
**See The Output**
<Tabs
groupId="engine-type"
- defaultValue="flink"
+ defaultValue="Zeta (local-mode)"
values={[
{label: 'Flink', value: 'flink'},
+ {label: 'Zeta (local-mode)', value: 'Zeta (local-mode)'},
+ {label: 'Zeta (cluster-mode)', value: 'Zeta (cluster-mode)'},
]}>
<TabItem value="flink">
@@ -282,6 +685,82 @@ To stop your job and delete your FlinkDeployment you can
simply:
kubectl delete -f seatunnel-flink.yaml
```
</TabItem>
+
+<TabItem value="Zeta (local-mode)">
+
+You may follow the logs of your job, after a successful startup (which can
take on the order of a minute in a fresh environment, seconds afterwards) you
can:
+
+```bash
+kubectl logs -f seatunnel
+```
+
+looks like the below (your content may be different since we use `FakeSource`
to automatically generate random stream data):
+
+```shell
+...
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25673: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: hRJdE, 1295862507
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25674: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: kXlew, 935460726
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25675: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: FrNOT, 1714358118
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25676: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: kSajX, 126709414
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25677: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: YhpQv, 2020198351
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25678: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: nApin, 691339553
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25679: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: KZNNa, 1720773736
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25680: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: uCUBI, 490868386
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25681: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: oTLmO, 98770781
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25682: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: UECud, 835494636
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25683: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: XNegY, 1602828896
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25684: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: LcFBx, 1400869177
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25685: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: EqSfF, 1933614060
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25686: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: BODIs, 1839533801
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25687: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: doxcI, 970104616
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25688: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: IEVYn, 371893767
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25689: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: YXYfq, 1719257882
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25690: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: LFWEm, 725033360
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25691: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: ypUrY, 1591744616
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25692: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: rlnzJ, 412162913
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25693: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: zWKnt, 976816261
+2023-10-07 08:20:12,797 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=0 rowIndex=25694: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT
: PXrsk, 43554541
+
+```
+
+To stop your job and delete your FlinkDeployment you can simply:
+
+```bash
+kubectl delete -f seatunnel.yaml
+```
+</TabItem>
+
+<TabItem value="Zeta (cluster-mode)">
+
+You may follow the logs of your job, after a successful startup (which can
take on the order of a minute in a fresh environment, seconds afterwards) you
can:
+
+```bash
+kubectl exec -it seatunnel-1 -- tail -f
/opt/seatunnel/logs/seatunnel-engine-server.log | grep ConsoleSinkWriter
+```
+
+looks like the below (your content may be different since we use `FakeSource`
to automatically generate random stream data):
+
+```shell
+...
+2023-10-10 08:05:07,283 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=7: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
IibHk, 820962465
+2023-10-10 08:05:07,283 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=8: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
lmKdb, 1072498088
+2023-10-10 08:05:07,283 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=9: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
iqGva, 918730371
+2023-10-10 08:05:07,284 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=10: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
JMHmq, 1130771733
+2023-10-10 08:05:07,284 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=11: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
rxoHF, 189596686
+2023-10-10 08:05:07,284 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=12: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
OSblw, 559472064
+2023-10-10 08:05:07,284 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=13: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
yTZjG, 1842482272
+2023-10-10 08:05:07,284 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=14: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
RRiMg, 1713777214
+2023-10-10 08:05:07,284 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=15: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
lRcsd, 1626041649
+2023-10-10 08:05:07,284 INFO
org.apache.seatunnel.connectors.seatunnel.console.sink.ConsoleSinkWriter -
subtaskIndex=1 rowIndex=16: SeaTunnelRow#tableId= SeaTunnelRow#kind=INSERT :
QrNNW, 41355294
+
+```
+
+To stop your job and delete your FlinkDeployment you can simply:
+
+```bash
+kubectl delete -f seatunnel-cluster.yaml
+```
+</TabItem>
</Tabs>
diff --git a/seatunnel-e2e/seatunnel-engine-e2e/pom.xml
b/seatunnel-e2e/seatunnel-engine-e2e/pom.xml
index 23c0f2d8dd..fe669d52b6 100644
--- a/seatunnel-e2e/seatunnel-engine-e2e/pom.xml
+++ b/seatunnel-e2e/seatunnel-engine-e2e/pom.xml
@@ -31,6 +31,7 @@
<modules>
<module>connector-seatunnel-e2e-base</module>
<module>connector-console-seatunnel-e2e</module>
+ <module>seatunnel-engine-k8s-e2e</module>
</modules>
<properties>
diff --git a/seatunnel-e2e/seatunnel-engine-e2e/pom.xml
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/pom.xml
similarity index 59%
copy from seatunnel-e2e/seatunnel-engine-e2e/pom.xml
copy to seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/pom.xml
index 23c0f2d8dd..da31270f91 100644
--- a/seatunnel-e2e/seatunnel-engine-e2e/pom.xml
+++ b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/pom.xml
@@ -20,64 +20,47 @@
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.seatunnel</groupId>
- <artifactId>seatunnel-e2e</artifactId>
+ <artifactId>seatunnel-engine-e2e</artifactId>
<version>${revision}</version>
</parent>
- <artifactId>seatunnel-engine-e2e</artifactId>
- <packaging>pom</packaging>
- <name>SeaTunnel : E2E : Engine :</name>
-
- <modules>
- <module>connector-seatunnel-e2e-base</module>
- <module>connector-console-seatunnel-e2e</module>
- </modules>
-
- <properties>
- <!-- SeaTunnel Engine use -->
- <hazelcast.version>5.1</hazelcast.version>
- </properties>
+ <artifactId>seatunnel-engine-k8s-e2e</artifactId>
+ <name>SeaTunnel : E2E : Engine : K8s</name>
<dependencies>
<dependency>
- <groupId>com.hazelcast</groupId>
- <artifactId>hazelcast</artifactId>
- <version>${hazelcast.version}</version>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-seatunnel-e2e-base</artifactId>
+ <version>${project.version}</version>
<classifier>tests</classifier>
+ <type>test-jar</type>
<scope>test</scope>
</dependency>
+ <!-- SeaTunnel connectors -->
<dependency>
<groupId>org.apache.seatunnel</groupId>
- <artifactId>seatunnel-e2e-common</artifactId>
+ <artifactId>connector-fake</artifactId>
<version>${project.version}</version>
- <type>test-jar</type>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.apache.seatunnel</groupId>
- <artifactId>seatunnel-engine-client</artifactId>
+ <artifactId>connector-console</artifactId>
<version>${project.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.seatunnel</groupId>
- <artifactId>connector-file-local</artifactId>
- </exclusion>
- <exclusion>
- <groupId>org.apache.seatunnel</groupId>
- <artifactId>connector-fake</artifactId>
- </exclusion>
- </exclusions>
+ <scope>test</scope>
</dependency>
<dependency>
- <groupId>org.apache.seatunnel</groupId>
- <artifactId>seatunnel-engine-server</artifactId>
- <version>${project.version}</version>
+ <groupId>io.kubernetes</groupId>
+ <artifactId>client-java</artifactId>
+ <version>16.0.0</version>
+ <scope>test</scope>
</dependency>
<dependency>
- <groupId>org.apache.seatunnel</groupId>
- <artifactId>seatunnel-starter</artifactId>
- <version>${project.version}</version>
+ <groupId>org.apache.maven</groupId>
+ <artifactId>maven-model</artifactId>
+ <version>3.6.0</version>
<scope>test</scope>
</dependency>
</dependencies>
+
</project>
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/java/org/apache/seatunnel/engine/e2e/k8s/KubernetesIT.java
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/java/org/apache/seatunnel/engine/e2e/k8s/KubernetesIT.java
new file mode 100644
index 0000000000..9d966ffe2f
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/java/org/apache/seatunnel/engine/e2e/k8s/KubernetesIT.java
@@ -0,0 +1,231 @@
+/*
+ * 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.seatunnel.engine.e2e.k8s;
+
+import org.apache.maven.model.Model;
+import org.apache.maven.model.io.xpp3.MavenXpp3Reader;
+
+import org.codehaus.plexus.util.FileUtils;
+import org.codehaus.plexus.util.xml.pull.XmlPullParserException;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.shaded.org.awaitility.Awaitility;
+
+import com.github.dockerjava.api.DockerClient;
+import com.github.dockerjava.api.command.BuildImageCmd;
+import com.github.dockerjava.api.model.Info;
+import io.kubernetes.client.openapi.ApiClient;
+import io.kubernetes.client.openapi.ApiException;
+import io.kubernetes.client.openapi.Configuration;
+import io.kubernetes.client.openapi.apis.AppsV1Api;
+import io.kubernetes.client.openapi.apis.CoreV1Api;
+import io.kubernetes.client.openapi.models.V1Service;
+import io.kubernetes.client.openapi.models.V1StatefulSet;
+import io.kubernetes.client.util.Config;
+import io.kubernetes.client.util.Yaml;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.File;
+import java.io.FileReader;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.util.Collections;
+import java.util.concurrent.TimeUnit;
+import java.util.regex.Pattern;
+
+import static
org.apache.seatunnel.e2e.common.util.ContainerUtil.PROJECT_ROOT_PATH;
+
+@Slf4j
+public class KubernetesIT {
+ private static final String namespace = "default";
+ private static final String svcName = "seatunnel";
+ private static final String stsName = "seatunnel";
+ private static final String podName = "seatunnel-0";
+
+ @Test
+ public void test()
+ throws IOException, XmlPullParserException, ApiException,
InterruptedException {
+ ApiClient client = Config.defaultClient();
+ AppsV1Api appsV1Api = new AppsV1Api(client);
+ CoreV1Api coreV1Api = new CoreV1Api(client);
+ DockerClient dockerClient = DockerClientFactory.lazyClient();
+ String targetPath =
+ PROJECT_ROOT_PATH
+ +
"/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources";
+ // If the Docker BaseDirectory is set as the root directory of the
project, the image
+ // created is too large, so choose to copy the files that need to be
created as images
+ // to the same level as the dockerfile.
+ String pomPath = PROJECT_ROOT_PATH + "/pom.xml";
+ MavenXpp3Reader pomReader = new MavenXpp3Reader();
+ Model model = pomReader.read(new FileReader(pomPath), true);
+ String artifactId = model.getArtifactId();
+ String tag = artifactId + ":latest";
+ Info info = dockerClient.infoCmd().exec();
+ log.info("Docker's environmental information");
+ log.info(info.toString());
+ if
(dockerClient.listImagesCmd().withImageNameFilter(tag).exec().isEmpty()) {
+ copyFileToCurrentResources(targetPath);
+ File file =
+ new File(
+ PROJECT_ROOT_PATH
+ +
"/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel_dockerfile");
+ BuildImageCmd buildImageCmd = dockerClient.buildImageCmd(file);
+ buildImageCmd.withTags(Collections.singleton(tag));
+ String imageId = buildImageCmd.start().awaitImageId();
+ Assertions.assertNotNull(imageId);
+ }
+ Configuration.setDefaultApiClient(client);
+ V1Service yamlSvc =
+ (V1Service)
+ Yaml.load(
+ new File(
+ PROJECT_ROOT_PATH
+ +
"/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel-service.yaml"));
+ V1StatefulSet yamlStatefulSet =
+ (V1StatefulSet)
+ Yaml.load(
+ new File(
+ PROJECT_ROOT_PATH
+ +
"/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel-statefulset.yaml"));
+ try {
+ coreV1Api.createNamespacedService(namespace, yamlSvc, null, null,
null, null);
+ appsV1Api.createNamespacedStatefulSet(
+ namespace, yamlStatefulSet, null, null, null, null);
+ Awaitility.await()
+ .atMost(30, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ V1StatefulSet v1StatefulSet =
+ appsV1Api.readNamespacedStatefulSet(
+ stsName, namespace, null);
+ Assertions.assertEquals(
+
v1StatefulSet.getStatus().getReadyReplicas(), 2);
+ });
+ // submit job
+ String command =
+ "sh /opt/seatunnel/bin/seatunnel.sh --config
/opt/seatunnel/config/v2.batch.config.template";
+ Process process =
+ Runtime.getRuntime()
+ .exec(
+ "kubectl exec -it "
+ + podName
+ + " -n "
+ + namespace
+ + " -- "
+ + command);
+ Assertions.assertEquals(0, process.waitFor());
+ // submit an error job
+ String commandError =
+ "sh /opt/seatunnel/bin/seatunnel.sh --config
/opt/seatunnel/config/v2.batch.config.template.error";
+ process =
+ Runtime.getRuntime()
+ .exec(
+ "kubectl exec -it "
+ + podName
+ + " -n "
+ + namespace
+ + " -- "
+ + commandError);
+ Assertions.assertEquals(1, process.waitFor());
+ } finally {
+ appsV1Api.deleteNamespacedStatefulSet(
+ stsName, namespace, null, null, null, null, null, null);
+ coreV1Api.deleteNamespacedService(
+ svcName, namespace, null, null, null, null, null, null);
+ }
+ }
+
+ private void copyFileToCurrentResources(String targetPath) throws
IOException {
+ File jarsPath = new File(targetPath + "/jars");
+ jarsPath.mkdirs();
+ File binPath = new File(targetPath + "/bin");
+ binPath.mkdirs();
+ File connectorsPath = new File(targetPath + "/connectors");
+ connectorsPath.mkdirs();
+ FileUtils.copyDirectory(
+ new File(PROJECT_ROOT_PATH + "/config"), new File(targetPath +
"/config"));
+ // replace hazelcast.yaml and hazelcast-client.yaml
+ Files.copy(
+ Paths.get(targetPath + "/custom_config/hazelcast.yaml"),
+ Paths.get(targetPath + "/config/hazelcast.yaml"),
+ StandardCopyOption.REPLACE_EXISTING);
+ Files.copy(
+ Paths.get(targetPath + "/custom_config/hazelcast-client.yaml"),
+ Paths.get(targetPath + "/config/hazelcast-client.yaml"),
+ StandardCopyOption.REPLACE_EXISTING);
+ Files.copy(
+ Paths.get(
+ PROJECT_ROOT_PATH
+ +
"/seatunnel-shade/seatunnel-hadoop3-3.1.4-uber/target/seatunnel-hadoop3-3.1.4-uber.jar"),
+ Paths.get(targetPath +
"/jars/seatunnel-hadoop3-3.1.4-uber.jar"),
+ StandardCopyOption.REPLACE_EXISTING);
+ Files.copy(
+ Paths.get(
+ PROJECT_ROOT_PATH
+ +
"/seatunnel-core/seatunnel-starter/target/seatunnel-starter.jar"),
+ Paths.get(targetPath + "/jars/seatunnel-starter.jar"),
+ StandardCopyOption.REPLACE_EXISTING);
+ Files.copy(
+ Paths.get(
+ PROJECT_ROOT_PATH
+ +
"/seatunnel-transforms-v2/target/seatunnel-transforms-v2.jar"),
+ Paths.get(targetPath + "/jars/seatunnel-transforms-v2.jar"),
+ StandardCopyOption.REPLACE_EXISTING);
+ Files.copy(
+ Paths.get(
+ PROJECT_ROOT_PATH
+ +
"/seatunnel-core/seatunnel-starter/src/main/bin/seatunnel.sh"),
+ Paths.get(targetPath + "/bin/seatunnel.sh"),
+ StandardCopyOption.REPLACE_EXISTING);
+ Files.copy(
+ Paths.get(
+ PROJECT_ROOT_PATH
+ +
"/seatunnel-core/seatunnel-starter/src/main/bin/seatunnel-cluster.sh"),
+ Paths.get(targetPath + "/bin/seatunnel-cluster.sh"),
+ StandardCopyOption.REPLACE_EXISTING);
+ Files.copy(
+ Paths.get(targetPath +
"/custom_config/plugin-mapping.properties"),
+ Paths.get(targetPath +
"/connectors/plugin-mapping.properties"),
+ StandardCopyOption.REPLACE_EXISTING);
+ fuzzyCopy(
+ PROJECT_ROOT_PATH +
"/seatunnel-connectors-v2/connector-fake/target/",
+ targetPath + "/connectors/",
+ "^connector-fake.*\\.jar$");
+ fuzzyCopy(
+ PROJECT_ROOT_PATH +
"/seatunnel-connectors-v2/connector-console/target/",
+ targetPath + "/connectors/",
+ "^connector-console.*\\.jar$");
+ }
+
+ private void fuzzyCopy(String sourceUrl, String targetUrl, String pattern)
throws IOException {
+ File dir = new File(sourceUrl);
+ File[] files = dir.listFiles();
+ Assertions.assertNotNull(files);
+ for (File file : files) {
+ if (Pattern.matches(pattern, file.getName())) {
+ Files.copy(
+ file.toPath(),
+ Paths.get(targetUrl + file.getName()),
+ StandardCopyOption.REPLACE_EXISTING);
+ }
+ }
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/custom_config/hazelcast-client.yaml
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/custom_config/hazelcast-client.yaml
new file mode 100644
index 0000000000..372a8673fb
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/custom_config/hazelcast-client.yaml
@@ -0,0 +1,23 @@
+#
+# 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.
+#
+
+hazelcast-client:
+ cluster-name: seatunnel
+ network:
+ cluster-members:
+ - seatunnel-0.seatunnel.default.svc.cluster.local:5801
+ - seatunnel-1.seatunnel.default.svc.cluster.local:5801
\ No newline at end of file
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/custom_config/hazelcast.yaml
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/custom_config/hazelcast.yaml
new file mode 100644
index 0000000000..10992ae39f
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/custom_config/hazelcast.yaml
@@ -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.
+#
+
+hazelcast:
+ cluster-name: seatunnel
+ network:
+ rest-api:
+ enabled: true
+ endpoint-groups:
+ CLUSTER_WRITE:
+ enabled: true
+ DATA:
+ enabled: true
+ join:
+ tcp-ip:
+ enabled: true
+ member-list:
+ - seatunnel-0.seatunnel.default.svc.cluster.local
+ - seatunnel-1.seatunnel.default.svc.cluster.local
+
+ port:
+ auto-increment: true
+ port-count: 100
+ port: 5801
+ properties:
+ hazelcast.invocation.max.retry.count: 100
+ hazelcast.invocation.retry.pause.millis: 1000
+ hazelcast.tcp.join.port.try.count: 30
+ hazelcast.slow.operation.detector.stacktrace.logging.enabled: true
+ hazelcast.logging.type: log4j2
+ hazelcast.operation.generic.thread.count: 200
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/custom_config/plugin-mapping.properties
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/custom_config/plugin-mapping.properties
new file mode 100644
index 0000000000..51a2db9025
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/custom_config/plugin-mapping.properties
@@ -0,0 +1,27 @@
+#
+# 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.
+#
+
+# This mapping is used to resolve the Jar package name without version (or
call artifactId)
+# corresponding to the module in the user Config, helping SeaTunnel to load
the correct Jar package.
+
+## *** WARNING **** : `seatunnel.source.XXX`, the `XXX` should be string which
SeaTunnelSource::getPluginName and TableSinkFactory::factoryIdentifier returned
value##
+
+# SeaTunnel Connector-V2
+
+seatunnel.source.FakeSource = connector-fake
+seatunnel.sink.Console = connector-console
+
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel-service.yaml
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel-service.yaml
new file mode 100644
index 0000000000..3817199500
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel-service.yaml
@@ -0,0 +1,29 @@
+#
+# 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.
+#
+
+
+apiVersion: v1
+kind: Service
+metadata:
+ name: seatunnel
+spec:
+ selector:
+ app: seatunnel
+ ports:
+ - port: 5801
+ name: seatunnel
+ clusterIP: None
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel-statefulset.yaml
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel-statefulset.yaml
new file mode 100644
index 0000000000..1ed5b6f504
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel-statefulset.yaml
@@ -0,0 +1,50 @@
+#
+# 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.
+#
+
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+ name: seatunnel
+spec:
+ serviceName: "seatunnel"
+ replicas: 2
+ selector:
+ matchLabels:
+ app: seatunnel
+ template:
+ metadata:
+ labels:
+ app: seatunnel
+ spec:
+ containers:
+ - name: seatunnel
+ image: seatunnel:latest
+ imagePullPolicy: IfNotPresent
+ ports:
+ - containerPort: 5801
+ name: client
+ command: [ 'sh' ]
+ args:
+ - "/opt/seatunnel/bin/seatunnel-cluster.sh"
+ - "-DJvmOption=-Xms2G -Xmx2G"
+ resources:
+ limits:
+ cpu: "1"
+ memory: 4G
+ requests:
+ cpu: "1"
+ memory: 2G
diff --git
a/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel_dockerfile
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel_dockerfile
new file mode 100644
index 0000000000..797f1ca044
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-engine-e2e/seatunnel-engine-k8s-e2e/src/test/resources/seatunnel_dockerfile
@@ -0,0 +1,30 @@
+#
+# 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 openjdk:8
+ENV SEATUNNEL_HOME="/opt/seatunnel"
+COPY /jars/seatunnel-hadoop3-3.1.4-uber.jar
${SEATUNNEL_HOME}/lib/seatunnel-hadoop3-3.1.4-uber.jar
+COPY /jars/seatunnel-transforms-v2.jar
${SEATUNNEL_HOME}/lib/sseatunnel-transforms-v2.jar
+COPY /jars/seatunnel-starter.jar
${SEATUNNEL_HOME}/starter/seatunnel-starter.jar
+COPY /bin ${SEATUNNEL_HOME}/bin
+COPY /connectors ${SEATUNNEL_HOME}/connectors
+COPY /config ${SEATUNNEL_HOME}/config
+RUN mkdir -p SEATUNNEL_HOME/logs
+WORKDIR /opt/seatunnel
+
+
+
