[feat-#1295][e2e]add ftp container

chunjun-e2e/src/test/java/com/dtstack/chunjun/connector/containers/ftp/SftpContainer.java

@@ -0,0 +1,58 @@
+/*
+ * 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 com.dtstack.chunjun.connector.containers.ftp;
+
+
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.WaitStrategy;
+import org.testcontainers.containers.wait.strategy.WaitStrategyTarget;
+import org.testcontainers.images.builder.ImageFromDockerfile;
+
+import java.net.URISyntaxException;
+import java.net.URL;
+import java.nio.file.Paths;
+import java.time.Duration;
+
+public class SftpContainer extends GenericContainer<SftpContainer> {
+
+    private static final URL SFTP_DOCKERFILE =
+            SftpContainer.class
+                    .getClassLoader()
+                    .getResource("docker/ftp/Dockerfile");
+
+    public SftpContainer(String imageName) throws URISyntaxException {
+        super(
+                new ImageFromDockerfile(imageName, true)
+                        .withDockerfile(Paths.get(SFTP_DOCKERFILE.toURI()))
+        );
+
+        waitingFor(
+                new WaitStrategy() {
+                    @Override
+                    public void waitUntilReady(WaitStrategyTarget waitStrategyTarget) {
+                    }
+
+                    @Override
+                    public WaitStrategy withStartupTimeout(Duration startupTimeout) {
+                        return null;
+                    }
+                });
+    }
+
+}

chunjun-e2e/src/test/java/com/dtstack/chunjun/connector/test/standalone/ftp/SftpSyncE2eITCase.java

@@ -0,0 +1,80 @@
+/*
+ * 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 com.dtstack.chunjun.connector.test.standalone.ftp;
+
+import com.dtstack.chunjun.connector.containers.ftp.SftpContainer;
+import com.dtstack.chunjun.connector.entity.JobAccumulatorResult;
+import com.dtstack.chunjun.connector.test.utils.ChunjunFlinkStandaloneTestEnvironment;
+
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+
+import java.net.URISyntaxException;
+import java.time.Duration;
+import java.util.stream.Stream;
+
+public class SftpSyncE2eITCase extends ChunjunFlinkStandaloneTestEnvironment {
+    private static final Logger LOG = LoggerFactory.getLogger(SftpSyncE2eITCase.class);
+
+    protected static final String sftpImageName = "ftp-e2e-stream";
+
+    protected SftpContainer sftpContainer;
+
+    private void initContainer() throws URISyntaxException {
+        sftpContainer = new SftpContainer(sftpImageName);
+        sftpContainer
+                .withNetwork(NETWORK)
+                .withNetworkAliases(sftpImageName)
+                .withLogConsumer(new Slf4jLogConsumer(LOG))
+                .dependsOn(flinkStandaloneContainer);
+    }
+
+    @Before
+    public void before() throws Exception {
+        super.before();
+        LOG.info("Starting sftp containers...");
+        initContainer();
+        Startables.deepStart(Stream.of(sftpContainer)).join();
+        Thread.sleep(5000);
+        LOG.info("sftp Containers are started.");
+    }
+
+    @After
+    public void after() {
+        super.after();
+        if (sftpContainer != null) {
+            sftpContainer.stop();
+        }
+    }
+
+    @Test
+    public void testFtpToStream() throws Exception {
+        submitSyncJobOnStandLone(
+                ChunjunFlinkStandaloneTestEnvironment.CHUNJUN_HOME
+                        + "/chunjun-examples/json/ftp/ftp_stream.json");
+        JobAccumulatorResult jobAccumulatorResult = waitUntilJobFinished(Duration.ofMinutes(30));
+        Assert.assertEquals(jobAccumulatorResult.getNumRead(), 20);
+    }
+}

chunjun-e2e/src/test/resources/docker/ftp/Dockerfile

@@ -0,0 +1,18 @@
+FROM ubuntu:16.04
+
+ADD test.csv /root/
+
+RUN apt-get update && apt-get install -y openssh-server
+RUN mkdir /var/run/sshd
+# Password & Authentication
+RUN echo 'root:admin123' | chpasswd
+RUN sed -i 's/PermitRootLogin prohibit-password/PermitRootLogin yes/' /etc/ssh/sshd_config
+
+# SSH & Keeping Session Alive
+RUN sed 's@session\s*required\s*pam_loginuid.so@session optional pam_loginuid.so@g' -i /etc/pam.d/sshd
+
+ENV NOTVISIBLE "in users profile"
+RUN echo "export VISIBLE=now" >> /etc/profile
+
+EXPOSE 22
+CMD ["/usr/sbin/sshd", "-D"]

chunjun-e2e/src/test/resources/docker/ftp/test.csv

@@ -0,0 +1,20 @@
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1
+1,1