This is an automated email from the ASF dual-hosted git repository.
oscerd pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-kafka-connector.git
The following commit(s) were added to refs/heads/main by this push:
new 253029081f Fix #743: add integration tests for the FTP, FTPS and SFTP
connectors (#1868)
253029081f is described below
commit 253029081f0be6a8bdb225f02563722b565ef3a4
Author: Andrea Cosentino <[email protected]>
AuthorDate: Thu Nov 5 13:09:04 2026 +0100
Fix #743: add integration tests for the FTP, FTPS and SFTP connectors
(#1868)
Add the tests/itests-ftp module with a source and a sink test for each of
the ftp, ftps and sftp connectors (#743, #744, #745), using the embedded
servers of camel-test-infra-ftp.
- Source tests drop a file into the polled directory and check the record
value and its CamelHeader.file header.
- Sink tests send records with a CamelHeader.file header and check the
files on the server.
- FTPS runs in explicit mode, the only one the ftps Kamelets support, and
validates the server certificate through camel.ssl.* with
camel.component.ftps.useGlobalSslContextParameters=true.
- SFTP checks the server host key, with strictHostKeyChecking=yes and
camel.component.sftp.knownHostsFile.
Signed-off-by: Andrea Cosentino <[email protected]>
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
tests/itests-ftp/pom.xml | 60 ++++++++++
.../kafkaconnector/ftp/common/FtpTestServers.java | 127 +++++++++++++++++++++
.../ftp/sink/CamelFtpPropertyFactory.java | 84 ++++++++++++++
.../ftp/sink/CamelSinkFtpITCase.java | 50 ++++++++
.../ftp/sink/CamelSinkFtpTestSupport.java | 118 +++++++++++++++++++
.../ftp/sink/CamelSinkFtpsITCase.java | 51 +++++++++
.../ftp/sink/CamelSinkSftpITCase.java | 53 +++++++++
.../ftp/source/CamelFtpPropertyFactory.java | 89 +++++++++++++++
.../ftp/source/CamelSourceFtpITCase.java | 50 ++++++++
.../ftp/source/CamelSourceFtpTestSupport.java | 101 ++++++++++++++++
.../ftp/source/CamelSourceFtpsITCase.java | 51 +++++++++
.../ftp/source/CamelSourceSftpITCase.java | 53 +++++++++
tests/pom.xml | 1 +
13 files changed, 888 insertions(+)
diff --git a/tests/itests-ftp/pom.xml b/tests/itests-ftp/pom.xml
new file mode 100644
index 0000000000..faa5a49f14
--- /dev/null
+++ b/tests/itests-ftp/pom.xml
@@ -0,0 +1,60 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+ ~ 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.
+ -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0";
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance";
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0
http://maven.apache.org/xsd/maven-4.0.0.xsd";>
+ <parent>
+ <groupId>org.apache.camel.kafkaconnector</groupId>
+ <artifactId>itests-parent</artifactId>
+ <version>4.22.1-SNAPSHOT</version>
+ <relativePath>../itests-parent/pom.xml</relativePath>
+ </parent>
+ <modelVersion>4.0.0</modelVersion>
+
+ <artifactId>itests-ftp</artifactId>
+ <name>Camel-Kafka-Connector :: Tests :: FTP</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.camel.kafkaconnector</groupId>
+ <artifactId>itests-common</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-ftp</artifactId>
+ </dependency>
+
+ <!-- Embedded FTP, FTPS and SFTP servers -->
+ <dependency>
+ <groupId>org.apache.camel</groupId>
+ <artifactId>camel-test-infra-ftp</artifactId>
+ <version>${camel.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.awaitility</groupId>
+ <artifactId>awaitility</artifactId>
+ <version>${awaitility-version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+</project>
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/common/FtpTestServers.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/common/FtpTestServers.java
new file mode 100644
index 0000000000..18d3b58e83
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/common/FtpTestServers.java
@@ -0,0 +1,127 @@
+/*
+ * 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.camel.kafkaconnector.ftp.common;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.UncheckedIOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.nio.file.StandardCopyOption;
+import java.util.Properties;
+
+import org.apache.camel.test.infra.ftp.services.FtpInfraService;
+import org.apache.camel.test.infra.ftp.services.embedded.EmbeddedConfiguration;
+import org.apache.camel.test.infra.ftp.services.embedded.FtpsEmbeddedService;
+
+/**
+ * Creates the embedded FTPS server used by the tests and provides the
connector configuration that verifies the
+ * identity of the FTPS and SFTP servers.
+ */
+public final class FtpTestServers {
+ // The key store bundled with camel-test-infra-ftp, holding the
self-signed certificate of the FTPS server. It is
+ // used from there rather than copied into this module, so that it follows
the Camel upgrades.
+ private static final String BUNDLED_KEY_STORE = "/server.jks";
+ private static final String KEY_STORE_PASSWORD = "password";
+ private static final Path KEY_STORE = Paths.get("target", "ftps",
"server.jks").toAbsolutePath();
+ private static final Path KNOWN_HOSTS = Paths.get("target", "sftp",
"known_hosts").toAbsolutePath();
+
+ private FtpTestServers() {
+
+ }
+
+ /**
+ * Creates an FTPS server running in explicit mode: the ftps Kamelets have
no option to use the implicit one.
+ *
+ * @return the FTPS service, to be registered as a JUnit extension
+ */
+ public static FtpsEmbeddedService createFtpsService() {
+ copyBundledKeyStore();
+
+ return new BundledKeyStoreFtpsService();
+ }
+
+ /**
+ * The connector configuration that makes the ftps component trust the
certificate of the FTPS server, through the
+ * global SSL configuration of the Camel context.
+ *
+ * @return the connector properties
+ */
+ public static Properties ftpsTrustProperties() {
+ final String keyStore = "file:" + KEY_STORE;
+
+ Properties properties = new Properties();
+ properties.put("camel.ssl.enabled", "true");
+ // Camel main only creates the global SSL context parameters when a
key store is configured
+ properties.put("camel.ssl.keyStore", keyStore);
+ properties.put("camel.ssl.keystorePassword", KEY_STORE_PASSWORD);
+ properties.put("camel.ssl.trustStore", keyStore);
+ properties.put("camel.ssl.trustStorePassword", KEY_STORE_PASSWORD);
+ properties.put("camel.component.ftps.useGlobalSslContextParameters",
"true");
+
+ return properties;
+ }
+
+ /**
+ * The connector configuration that makes the sftp component verify the
host key of the SFTP server against a
+ * known_hosts file. The sftp Kamelets have no option for it, so it is set
on the component.
+ *
+ * @param service the running SFTP service
+ * @return the connector properties
+ */
+ public static Properties sftpKnownHostsProperties(FtpInfraService service)
{
+ try {
+ Files.createDirectories(KNOWN_HOSTS.getParent());
+ Files.writeString(KNOWN_HOSTS, service.knownHostsEntry() +
System.lineSeparator());
+ } catch (IOException e) {
+ throw new UncheckedIOException("Unable to write the known hosts
file " + KNOWN_HOSTS, e);
+ }
+
+ Properties properties = new Properties();
+ properties.put("camel.component.sftp.knownHostsFile",
KNOWN_HOSTS.toString());
+
+ return properties;
+ }
+
+ private static void copyBundledKeyStore() {
+ try (InputStream keyStore =
FtpsEmbeddedService.class.getResourceAsStream(BUNDLED_KEY_STORE)) {
+ if (keyStore == null) {
+ throw new IllegalStateException("The key store bundled with
camel-test-infra-ftp was not found");
+ }
+
+ Files.createDirectories(KEY_STORE.getParent());
+ Files.copy(keyStore, KEY_STORE,
StandardCopyOption.REPLACE_EXISTING);
+ } catch (IOException e) {
+ throw new UncheckedIOException("Unable to copy the FTPS key store
to " + KEY_STORE, e);
+ }
+ }
+
+ /**
+ * FtpsEmbeddedService only looks for its key store at
./src/test/resources/server.jks, so it is pointed to the copy
+ * of the bundled one instead.
+ */
+ private static final class BundledKeyStoreFtpsService extends
FtpsEmbeddedService {
+ BundledKeyStoreFtpsService() {
+ // explicit mode, TLSv1.2 (as in the Camel FTPS tests using this
server) and no client authentication
+ super(new EmbeddedConfiguration.SecurityConfiguration(false,
"TLSv1.2", false));
+
+ embeddedConfigurationTemplate.withKeyStore(KEY_STORE.toString());
+ }
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelFtpPropertyFactory.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelFtpPropertyFactory.java
new file mode 100644
index 0000000000..514ce583e1
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelFtpPropertyFactory.java
@@ -0,0 +1,84 @@
+/*
+ * 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.camel.kafkaconnector.ftp.sink;
+
+import org.apache.camel.kafkaconnector.common.SinkConnectorPropertyFactory;
+
+/**
+ * Property factory for the ftp-sink, ftps-sink and sftp-sink connectors,
whose Kamelets share the same options.
+ */
+final class CamelFtpPropertyFactory extends
SinkConnectorPropertyFactory<CamelFtpPropertyFactory> {
+ private final String kamelet;
+
+ private CamelFtpPropertyFactory(String kamelet) {
+ this.kamelet = kamelet;
+ }
+
+ public CamelFtpPropertyFactory withConnectionHost(String connectionHost) {
+ return withKameletProperty("connectionHost", connectionHost);
+ }
+
+ public CamelFtpPropertyFactory withConnectionPort(int connectionPort) {
+ return withKameletProperty("connectionPort",
String.valueOf(connectionPort));
+ }
+
+ public CamelFtpPropertyFactory withUsername(String username) {
+ return withKameletProperty("username", username);
+ }
+
+ public CamelFtpPropertyFactory withPassword(String password) {
+ return withKameletProperty("password", password);
+ }
+
+ public CamelFtpPropertyFactory withDirectoryName(String directoryName) {
+ return withKameletProperty("directoryName", directoryName);
+ }
+
+ public CamelFtpPropertyFactory withStrictHostKeyChecking(String
strictHostKeyChecking) {
+ return withKameletProperty("strictHostKeyChecking",
strictHostKeyChecking);
+ }
+
+ private CamelFtpPropertyFactory withKameletProperty(String name, String
value) {
+ return setProperty("camel.kamelet." + kamelet + "." + name, value);
+ }
+
+ public static CamelFtpPropertyFactory ftp() {
+ return basic("ftp-sink", "CamelFtpSinkConnector",
+
"org.apache.camel.kafkaconnector.ftpsink.CamelFtpsinkSinkConnector");
+ }
+
+ public static CamelFtpPropertyFactory ftps() {
+ return basic("ftps-sink", "CamelFtpsSinkConnector",
+
"org.apache.camel.kafkaconnector.ftpssink.CamelFtpssinkSinkConnector");
+ }
+
+ public static CamelFtpPropertyFactory sftp() {
+ return basic("sftp-sink", "CamelSftpSinkConnector",
+
"org.apache.camel.kafkaconnector.sftpsink.CamelSftpsinkSinkConnector");
+ }
+
+ private static CamelFtpPropertyFactory basic(String kamelet, String name,
String connectorClass) {
+ return new CamelFtpPropertyFactory(kamelet)
+ .withName(name)
+ .withTasksMax(1)
+ .withConnectorClass(connectorClass)
+
.withKeyConverterClass("org.apache.kafka.connect.storage.StringConverter")
+
.withValueConverterClass("org.apache.kafka.connect.storage.StringConverter")
+ .setProperty("camel.component.kamelet.location", "kamelets");
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkFtpITCase.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkFtpITCase.java
new file mode 100644
index 0000000000..f4849922df
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkFtpITCase.java
@@ -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.
+ */
+
+package org.apache.camel.kafkaconnector.ftp.sink;
+
+import java.nio.file.Path;
+
+import org.apache.camel.test.infra.ftp.services.FtpService;
+import org.apache.camel.test.infra.ftp.services.FtpServiceFactory;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class CamelSinkFtpITCase extends CamelSinkFtpTestSupport {
+ @RegisterExtension
+ public static FtpService service =
FtpServiceFactory.createEmbeddedService();
+
+ @Override
+ protected String[] getConnectorsInTest() {
+ return new String[] {"camel-ftp-sink-kafka-connector"};
+ }
+
+ @Override
+ protected Path localDirectory() {
+ // the FTP user home is the root directory of the server
+ return service.getFtpRootDir().resolve(DIRECTORY);
+ }
+
+ @Override
+ protected CamelFtpPropertyFactory connectorPropertyFactory() {
+ return CamelFtpPropertyFactory.ftp()
+ .withConnectionHost(service.host())
+ .withConnectionPort(service.port())
+ .withUsername(service.username())
+ .withPassword(service.password())
+ .withDirectoryName(DIRECTORY);
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkFtpTestSupport.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkFtpTestSupport.java
new file mode 100644
index 0000000000..7296fc38d1
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkFtpTestSupport.java
@@ -0,0 +1,118 @@
+/*
+ * 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.camel.kafkaconnector.ftp.sink;
+
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+import org.apache.camel.kafkaconnector.CamelSinkTask;
+import org.apache.camel.kafkaconnector.common.test.CamelSinkTestSupport;
+import org.apache.camel.kafkaconnector.common.test.StringMessageProducer;
+import org.awaitility.Awaitility;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Sends records to the sink connector and checks the files written on the
server. The name of each file comes from
+ * the CamelHeader.file Kafka header of the record, which the connector maps
to the file header of the Kamelet.
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public abstract class CamelSinkFtpTestSupport extends CamelSinkTestSupport {
+ protected static final String DIRECTORY = "sink";
+
+ private static final int EXPECT = 3;
+
+ private String topic;
+
+ private static class FileNameHeaderProducer extends StringMessageProducer {
+ FileNameHeaderProducer(String bootstrapServer, String topicName, int
count) {
+ super(bootstrapServer, topicName, count);
+ }
+
+ @Override
+ public Map<String, String> messageHeaders(String text, int current) {
+ return Collections.singletonMap(CamelSinkTask.HEADER_CAMEL_PREFIX
+ "file", fileName(current));
+ }
+
+ @Override
+ public String testMessageContent(int current) {
+ return fileContent(current);
+ }
+ }
+
+ /**
+ * @return the local path of the directory the connector writes the files
to
+ */
+ protected abstract Path localDirectory();
+
+ /**
+ * @return the connector configuration for the server under test, without
the Kafka topic
+ */
+ protected abstract CamelFtpPropertyFactory connectorPropertyFactory();
+
+ @BeforeEach
+ public void setUp() {
+ topic = getTopicForTest(this);
+ }
+
+ @Override
+ protected void consumeMessages(CountDownLatch latch) {
+ // nothing to consume from Kafka: verifyMessages checks the files
written on the server
+ latch.countDown();
+ }
+
+ @Override
+ protected void verifyMessages(CountDownLatch latch) {
+ final Path directory = localDirectory();
+
+ Awaitility.await()
+ .atMost(Duration.ofSeconds(60))
+ .untilAsserted(() -> {
+ for (int i = 0; i < EXPECT; i++) {
+ Path file = directory.resolve(fileName(i));
+
+ assertTrue(Files.exists(file), "The file " + file + "
was not written");
+ assertEquals(fileContent(i), Files.readString(file),
"Unexpected content in " + file);
+ }
+ });
+ }
+
+ @Test
+ @Timeout(180)
+ public void testStoreFiles() throws Exception {
+ runTest(connectorPropertyFactory().withTopics(topic),
+ new
FileNameHeaderProducer(getKafkaService().getBootstrapServers(), topic, EXPECT));
+ }
+
+ private static String fileName(int current) {
+ return "sink-test-" + current + ".txt";
+ }
+
+ private static String fileContent(int current) {
+ return "Camel Kafka Connector sink test " + current;
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkFtpsITCase.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkFtpsITCase.java
new file mode 100644
index 0000000000..437a9eac5c
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkFtpsITCase.java
@@ -0,0 +1,51 @@
+/*
+ * 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.camel.kafkaconnector.ftp.sink;
+
+import java.nio.file.Path;
+
+import org.apache.camel.kafkaconnector.ftp.common.FtpTestServers;
+import org.apache.camel.test.infra.ftp.services.FtpService;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class CamelSinkFtpsITCase extends CamelSinkFtpTestSupport {
+ @RegisterExtension
+ public static FtpService service = FtpTestServers.createFtpsService();
+
+ @Override
+ protected String[] getConnectorsInTest() {
+ return new String[] {"camel-ftps-sink-kafka-connector"};
+ }
+
+ @Override
+ protected Path localDirectory() {
+ // the FTPS user home is the root directory of the server
+ return service.getFtpRootDir().resolve(DIRECTORY);
+ }
+
+ @Override
+ protected CamelFtpPropertyFactory connectorPropertyFactory() {
+ return CamelFtpPropertyFactory.ftps()
+ .withConnectionHost(service.host())
+ .withConnectionPort(service.port())
+ .withUsername(service.username())
+ .withPassword(service.password())
+ .withDirectoryName(DIRECTORY)
+ .merge(FtpTestServers.ftpsTrustProperties());
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkSftpITCase.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkSftpITCase.java
new file mode 100644
index 0000000000..3590ebe6e3
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/sink/CamelSinkSftpITCase.java
@@ -0,0 +1,53 @@
+/*
+ * 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.camel.kafkaconnector.ftp.sink;
+
+import java.nio.file.Path;
+
+import org.apache.camel.kafkaconnector.ftp.common.FtpTestServers;
+import org.apache.camel.test.infra.ftp.services.FtpService;
+import org.apache.camel.test.infra.ftp.services.embedded.SftpEmbeddedService;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class CamelSinkSftpITCase extends CamelSinkFtpTestSupport {
+ @RegisterExtension
+ public static FtpService service = new SftpEmbeddedService();
+
+ @Override
+ protected String[] getConnectorsInTest() {
+ return new String[] {"camel-sftp-sink-kafka-connector"};
+ }
+
+ @Override
+ protected Path localDirectory() {
+ // the SFTP server exposes the local file system: its root directory
is relative to the working directory
+ return service.getFtpRootDir().resolve(DIRECTORY);
+ }
+
+ @Override
+ protected CamelFtpPropertyFactory connectorPropertyFactory() {
+ return CamelFtpPropertyFactory.sftp()
+ .withConnectionHost(service.host())
+ .withConnectionPort(service.port())
+ .withUsername(service.username())
+ .withPassword(service.password())
+ .withDirectoryName(localDirectory().toString())
+ .withStrictHostKeyChecking("yes")
+ .merge(FtpTestServers.sftpKnownHostsProperties(service));
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelFtpPropertyFactory.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelFtpPropertyFactory.java
new file mode 100644
index 0000000000..d911cf6826
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelFtpPropertyFactory.java
@@ -0,0 +1,89 @@
+/*
+ * 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.camel.kafkaconnector.ftp.source;
+
+import org.apache.camel.kafkaconnector.common.SourceConnectorPropertyFactory;
+
+/**
+ * Property factory for the ftp-source, ftps-source and sftp-source
connectors, whose Kamelets share the same options.
+ */
+final class CamelFtpPropertyFactory extends
SourceConnectorPropertyFactory<CamelFtpPropertyFactory> {
+ private final String kamelet;
+
+ private CamelFtpPropertyFactory(String kamelet) {
+ this.kamelet = kamelet;
+ }
+
+ public CamelFtpPropertyFactory withConnectionHost(String connectionHost) {
+ return withKameletProperty("connectionHost", connectionHost);
+ }
+
+ public CamelFtpPropertyFactory withConnectionPort(int connectionPort) {
+ return withKameletProperty("connectionPort",
String.valueOf(connectionPort));
+ }
+
+ public CamelFtpPropertyFactory withUsername(String username) {
+ return withKameletProperty("username", username);
+ }
+
+ public CamelFtpPropertyFactory withPassword(String password) {
+ return withKameletProperty("password", password);
+ }
+
+ public CamelFtpPropertyFactory withDirectoryName(String directoryName) {
+ return withKameletProperty("directoryName", directoryName);
+ }
+
+ public CamelFtpPropertyFactory withStrictHostKeyChecking(String
strictHostKeyChecking) {
+ return withKameletProperty("strictHostKeyChecking",
strictHostKeyChecking);
+ }
+
+ private CamelFtpPropertyFactory withKameletProperty(String name, String
value) {
+ return setProperty("camel.kamelet." + kamelet + "." + name, value);
+ }
+
+ public static CamelFtpPropertyFactory ftp() {
+ return basic("ftp-source", "CamelFtpSourceConnector",
+
"org.apache.camel.kafkaconnector.ftpsource.CamelFtpsourceSourceConnector");
+ }
+
+ public static CamelFtpPropertyFactory ftps() {
+ return basic("ftps-source", "CamelFtpsSourceConnector",
+
"org.apache.camel.kafkaconnector.ftpssource.CamelFtpssourceSourceConnector");
+ }
+
+ public static CamelFtpPropertyFactory sftp() {
+ return basic("sftp-source", "CamelSftpSourceConnector",
+
"org.apache.camel.kafkaconnector.sftpsource.CamelSftpsourceSourceConnector");
+ }
+
+ private static CamelFtpPropertyFactory basic(String kamelet, String name,
String connectorClass) {
+ return new CamelFtpPropertyFactory(kamelet)
+ .withName(name)
+ .withTasksMax(1)
+ .withConnectorClass(connectorClass)
+
.withKeyConverterClass("org.apache.kafka.connect.storage.StringConverter")
+
.withValueConverterClass("org.apache.kafka.connect.storage.StringConverter")
+ .setProperty("camel.component.kamelet.location", "kamelets")
+ // the Kamelets hand over the file content as a stream:
convert it with the Camel type converters
+ .withTransformsConfig("FileContentToString")
+ .withEntry("type",
"org.apache.camel.kafkaconnector.transforms.CamelTypeConverterTransform$Value")
+ .withEntry("target.type", "java.lang.String")
+ .end();
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceFtpITCase.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceFtpITCase.java
new file mode 100644
index 0000000000..1657ae45f0
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceFtpITCase.java
@@ -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.
+ */
+
+package org.apache.camel.kafkaconnector.ftp.source;
+
+import java.nio.file.Path;
+
+import org.apache.camel.test.infra.ftp.services.FtpService;
+import org.apache.camel.test.infra.ftp.services.FtpServiceFactory;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class CamelSourceFtpITCase extends CamelSourceFtpTestSupport {
+ @RegisterExtension
+ public static FtpService service =
FtpServiceFactory.createEmbeddedService();
+
+ @Override
+ protected String[] getConnectorsInTest() {
+ return new String[] {"camel-ftp-source-kafka-connector"};
+ }
+
+ @Override
+ protected Path localDirectory() {
+ // the FTP user home is the root directory of the server
+ return service.getFtpRootDir().resolve(DIRECTORY);
+ }
+
+ @Override
+ protected CamelFtpPropertyFactory connectorPropertyFactory() {
+ return CamelFtpPropertyFactory.ftp()
+ .withConnectionHost(service.host())
+ .withConnectionPort(service.port())
+ .withUsername(service.username())
+ .withPassword(service.password())
+ .withDirectoryName(DIRECTORY);
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceFtpTestSupport.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceFtpTestSupport.java
new file mode 100644
index 0000000000..4bef7b7e6e
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceFtpTestSupport.java
@@ -0,0 +1,101 @@
+/*
+ * 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.camel.kafkaconnector.ftp.source;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardCopyOption;
+import java.util.List;
+import java.util.concurrent.ExecutionException;
+
+import org.apache.camel.kafkaconnector.CamelSourceTask;
+import org.apache.camel.kafkaconnector.common.test.CamelSourceTestSupport;
+import org.apache.camel.kafkaconnector.common.test.TestMessageConsumer;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.header.Header;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.TestInstance;
+import org.junit.jupiter.api.Timeout;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Drops a file in the directory polled by the source connector and checks the
record sent to Kafka.
+ */
+@TestInstance(TestInstance.Lifecycle.PER_CLASS)
+public abstract class CamelSourceFtpTestSupport extends CamelSourceTestSupport
{
+ protected static final String DIRECTORY = "source";
+
+ private static final String FILE_NAME = "source-test.txt";
+ private static final String FILE_CONTENT = "Camel Kafka Connector source
test";
+
+ private String topic;
+
+ /**
+ * @return the local path of the directory polled by the connector
+ */
+ protected abstract Path localDirectory();
+
+ /**
+ * @return the connector configuration for the server under test, without
the Kafka topic
+ */
+ protected abstract CamelFtpPropertyFactory connectorPropertyFactory();
+
+ @BeforeEach
+ public void setUp() {
+ topic = getTopicForTest(this);
+ }
+
+ @Override
+ protected void produceTestData() {
+ try {
+ Path directory = localDirectory();
+ Files.createDirectories(directory);
+
+ // write the file aside and move it in, so that the connector
cannot pick it up partially written
+ Path file = Files.writeString(directory.resolveSibling(FILE_NAME),
FILE_CONTENT);
+ Files.move(file, directory.resolve(FILE_NAME),
StandardCopyOption.ATOMIC_MOVE);
+ } catch (IOException e) {
+ fail("Unable to create the test file: " + e.getMessage());
+ }
+ }
+
+ @Override
+ protected void verifyMessages(TestMessageConsumer<?> consumer) {
+ List<? extends ConsumerRecord<String, ?>> records =
consumer.consumedMessages();
+ assertEquals(1, records.size(), "Did not receive the expected number
of records");
+
+ ConsumerRecord<String, ?> record = records.get(0);
+ assertEquals(FILE_CONTENT, record.value(), "The record value is not
the content of the file");
+
+ Header fileName =
record.headers().lastHeader(CamelSourceTask.HEADER_CAMEL_PREFIX + "file");
+ assertNotNull(fileName, "The record does not carry the file name
header");
+ assertEquals(FILE_NAME, new String(fileName.value(),
StandardCharsets.UTF_8), "Unexpected file name header");
+ }
+
+ @Test
+ @Timeout(180)
+ public void testRetrieveFile() throws ExecutionException,
InterruptedException {
+ runTest(connectorPropertyFactory().withKafkaTopic(topic), topic, 1);
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceFtpsITCase.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceFtpsITCase.java
new file mode 100644
index 0000000000..59035641ab
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceFtpsITCase.java
@@ -0,0 +1,51 @@
+/*
+ * 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.camel.kafkaconnector.ftp.source;
+
+import java.nio.file.Path;
+
+import org.apache.camel.kafkaconnector.ftp.common.FtpTestServers;
+import org.apache.camel.test.infra.ftp.services.FtpService;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class CamelSourceFtpsITCase extends CamelSourceFtpTestSupport {
+ @RegisterExtension
+ public static FtpService service = FtpTestServers.createFtpsService();
+
+ @Override
+ protected String[] getConnectorsInTest() {
+ return new String[] {"camel-ftps-source-kafka-connector"};
+ }
+
+ @Override
+ protected Path localDirectory() {
+ // the FTPS user home is the root directory of the server
+ return service.getFtpRootDir().resolve(DIRECTORY);
+ }
+
+ @Override
+ protected CamelFtpPropertyFactory connectorPropertyFactory() {
+ return CamelFtpPropertyFactory.ftps()
+ .withConnectionHost(service.host())
+ .withConnectionPort(service.port())
+ .withUsername(service.username())
+ .withPassword(service.password())
+ .withDirectoryName(DIRECTORY)
+ .merge(FtpTestServers.ftpsTrustProperties());
+ }
+}
diff --git
a/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceSftpITCase.java
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceSftpITCase.java
new file mode 100644
index 0000000000..2e81c7466e
--- /dev/null
+++
b/tests/itests-ftp/src/test/java/org/apache/camel/kafkaconnector/ftp/source/CamelSourceSftpITCase.java
@@ -0,0 +1,53 @@
+/*
+ * 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.camel.kafkaconnector.ftp.source;
+
+import java.nio.file.Path;
+
+import org.apache.camel.kafkaconnector.ftp.common.FtpTestServers;
+import org.apache.camel.test.infra.ftp.services.FtpService;
+import org.apache.camel.test.infra.ftp.services.embedded.SftpEmbeddedService;
+import org.junit.jupiter.api.extension.RegisterExtension;
+
+public class CamelSourceSftpITCase extends CamelSourceFtpTestSupport {
+ @RegisterExtension
+ public static FtpService service = new SftpEmbeddedService();
+
+ @Override
+ protected String[] getConnectorsInTest() {
+ return new String[] {"camel-sftp-source-kafka-connector"};
+ }
+
+ @Override
+ protected Path localDirectory() {
+ // the SFTP server exposes the local file system: its root directory
is relative to the working directory
+ return service.getFtpRootDir().resolve(DIRECTORY);
+ }
+
+ @Override
+ protected CamelFtpPropertyFactory connectorPropertyFactory() {
+ return CamelFtpPropertyFactory.sftp()
+ .withConnectionHost(service.host())
+ .withConnectionPort(service.port())
+ .withUsername(service.username())
+ .withPassword(service.password())
+ .withDirectoryName(localDirectory().toString())
+ .withStrictHostKeyChecking("yes")
+ .merge(FtpTestServers.sftpKnownHostsProperties(service));
+ }
+}
diff --git a/tests/pom.xml b/tests/pom.xml
index 14f6e5b4ae..616b8cc942 100644
--- a/tests/pom.xml
+++ b/tests/pom.xml
@@ -48,6 +48,7 @@
<module>itests-elasticsearch</module>
<module>itests-syslog</module>
<module>itests-file</module>
+ <module>itests-ftp</module>
<module>itests-http</module>
<module>itests-https</module>
<module>itests-timer</module>
