<a id="sftp-source-connector"></a>

# SFTP Source Connector for Confluent Platform

The SFTP Source connector package for Kafka Connect watches a directory on an
SSH File Transfer Protocol (SFTP) server and reads new files as they arrive.
Each record in a file is converted using a user-supplied schema or an
auto-generated schema.

After the connector reads a file, it moves the file to
the directory configured in `finished.path`.

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

The following connectors are included in the SFTP Source connector package:

* [CSV Source Connector for Confluent Platform](csv_source_connector.md#sftp-csv-source-connector)
* [JSON Source Connector for Confluent Platform](json_source_connector.md#sftp-json-source-connector)
* [Schemaless JSON Source Connector for Confluent Platform](schemaless_json_source_connector.md#sftp-schemaless-json-source-connector)
* [Binary File Source Connector for Confluent Platform](binary_source_connector.md#sftp-binary-file-source-connector)

## SFTP Source connector features

The SFTP Source connectors include the following features:

- [At least once delivery](#sftp-source-at-least-once-delivery)
- [Supports one task](#sftp-source-supports-one-task)
- [Client-side encryption](#sftp-source-csfle-sm)

<a id="sftp-source-at-least-once-delivery"></a>

### At least once delivery

Each connector in the SFTP Source connector package guarantees that records are
delivered at least once to the Kafka topic. If the connector restarts, there may
be some duplicate records in the Kafka topic.

<a id="sftp-source-supports-one-task"></a>

### Supports one task

The connectors in the SFTP Source connector package each support running only
one task.

<a id="sftp-source-csfle-sm"></a>

### Client-side encryption

This connector supports Client-Side Field Level Encryption (CSFLE) and Client-Side Payload Encryption (CSPE). For more information, see [Manage Client-Side Encryption](https://docs.confluent.io/platform/current/connect/manage-csfle.html).

## File metadata

<a id="sftp-file-metadata-headers"></a>

Starting version 2.3.0, the SFTP Source connector supports exporting file metadata in Kafka headers.
The following information is stored in each record’s headers:

* `file.path` - The path of the source file on the SFTP server.
* `file.name` - The name of the file from which the record was read.
* `file.name.without.extension` - The name of the file without the extension.
* `file.last.modified` - The last modification time of the file in seconds.
* `file.length` - The size of the file in bytes.
* `file.offset` - The offset of the record in the file. (N/A for Binary files)

## License

You can use this connector for a 30-day trial period without a license key.

After 30 days, you must purchase a connector subscription which includes [Confluent enterprise license](/platform/current/installation/license.html#enterprise-subscription-license) keys to subscribers, along with [enterprise-level support](https://www.confluent.io/subscription/) for Confluent Platform and your connectors. If you are a subscriber, you can contact [Confluent Support](https://support.confluent.io/) for more information.

For license properties, see [Confluent Platform license](../sink-connector/configuration_options.md#sftp-sink-connector-license-config). For
information about the license topic, see
[License topic configuration](../sink-connector/configuration_options.md#sftp-sink-license-topic-configuration).

## Limitations

The SFTP Source connector has the following limitations:

- Currently, the SFTP Source connector reads and moves a file only once while
  processing files from a specific SFTP directory. If a file with the same
  name is produced later in the same SFTP directory, the connector does not
  process it.
- When the `schema.generation.enabled` configuration is enabled, the
  connector offers limited schema detection capabilities and does not support
  nested or array schemas.

## SFTP Source connector configuration properties

For a list of configuration properties for the SFTP Source connectors, see the
specific connector documentation.

For an example of how to get Kafka Connect connected to [Confluent Cloud](/cloud/current/index.html), see
[Connect Self-Managed Kafka Connect to Confluent Cloud](/cloud/current/cp-component/connect-cloud-config.html#distributed-cluster).

## Install the SFTP Source connector

You can install this connector by using the [confluent connect
plugin
install](https://docs.confluent.io/confluent-cli/current/command-reference/connect/plugin/confluent_connect_plugin_install.html)
command, or by manually downloading the ZIP file.

### Prerequisites

- You must install the connector on every machine where Connect will run.
- If you want to install the connector using Confluent Marketplace, you must install
  the [Confluent Hub Client](https://docs.confluent.io/home/connect/confluent-hub/client.html). This
  is installed by default with Confluent Enterprise.
- Kafka Broker: Confluent Platform 3.3.0 or later, or Kafka 0.11.0 or later.
- Connect: Confluent Platform 4.0.0 or later, or Kafka 1.0.0 or later.
- Java 1.8.

### Install the connector using the Confluent CLI

To install the `latest` connector version using [Confluent Hub Client](https://www.confluent.io/hub/), navigate to your Confluent Platform installation directory
and run the following command:

```bash
confluent connect plugin install confluentinc/kafka-connect-sftp:latest
```

You can install a specific version by replacing `latest` with a version number
as shown in the following example:

```bash
confluent connect plugin install confluentinc/kafka-connect-sftp:3.2.0
```

### Install the connector manually

[Download and extract the ZIP file](https://www.confluent.io/hub/confluentinc/kafka-connect-sftp) for your
connector and then follow the manual connector installation
[instructions](/kafka-connectors/self-managed/install.html).

<a id="sftp-source-connector-quickstart"></a>

## Quick start

The following steps show the `SftpCsvSourceConnector` loading a mock CSV file
to a Kafka topic named `sftp-testing-topic`. The other connectors are similar
but load from different file types.

1. Install the connector through the [Confluent Hub
   Client](/kafka-connectors/self-managed/confluent-hub/client.html).
   ```bash
   # run from your Confluent Platform installation directory
   confluent connect plugin install confluentinc/kafka-connect-sftp:latest
   ```
2. Start Confluent Platform.
   ```bash
   confluent local start
   ```
3. Set up an SFTP data directory for files to process, generate test data
   locally, and push it to the SFTP server.
   ```bash
   echo $'id,first_name,last_name,email,gender,ip_address,last_login,account_balance,country,favorite_color\n1,Salmon,Baitman,sbaitman0@feedburner.com,Male,120.181.75.98,2015-03-01T06:01:15Z,17462.66,IT,#f09bc0\n2,Debby,Brea,dbrea1@icio.us,Female,153.239.187.49,2018-10-21T12:27:12Z,14693.49,CZ,#73893a' > "csv-sftp-source.csv"
   ```
4. Set up SFTP directories for files with errors and files that finished
   successfully.
   ```bash
   mkdir error && mkdir finished
   ```
5. Create a `sftp.json` file with the following contents:
   ```json
   {
     "name": "CsvSFTP",
     "config": {
       "tasks.max": "1",
       "connector.class": "io.confluent.connect.sftp.SftpCsvSourceConnector",
       "cleanup.policy":"NONE",
       "behavior.on.error":"IGNORE",
       "input.path": "/path/to/data",
       "error.path": "/path/to/error",
       "finished.path": "/path/to/finished",
       "input.file.pattern": "csv-sftp-source.csv",
       "sftp.username":"username",
       "sftp.password":"password",
       "sftp.host":"localhost",
       "sftp.port":"22",
       "kafka.topic": "sftp-testing-topic",
       "csv.first.row.as.header": "true",
       "schema.generation.enabled": "true"
     }
   }
   ```
6. Load the SFTP CSV Source connector.
   ```bash
   confluent local load CsvSFTP --config sftp.json
   ```

   #### IMPORTANT
   Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html)
   in production environments.
7. Confirm that the connector is in a `RUNNING` state.
   ```bash
   confluent local status CsvSFTP
   ```
8. Confirm that the messages are being sent to Kafka.
   ```bash
   kafka-avro-console-consumer \
       --bootstrap-server localhost:9092 \
       --property schema.registry.url=http://localhost:8081 \
       --topic sftp-testing-topic \
       --from-beginning | jq '.'
   ```
9. If you have opted for `"cleanup.policy":"MOVE"`, confirm that the source
   CSV file has been moved to the `finished` directory.

<a id="sftp-connector-examples"></a>

## Examples

* [CSV with Schema Example](csv_source_connector.md#sftp-connector-csv-with-schema-example)
* [TSV Input File Example](csv_source_connector.md#sftp-connector-tsv-example)
* [JSON Schemaless Source Connector Example](schemaless_json_source_connector.md#sftp-schemaless-json-source-connector-example)
* [JSON Source Connector Example](json_source_connector.md#sftp-connector-json-example)
* [Binary File Source Connector Example](binary_source_connector.md#sftp-binary-file-source-connector-example)