<a id="s3source-connector"></a>

# Backup and Restore Amazon S3 Source Connector for Confluent Platform

The Backup and Restore Kafka Connect  Amazon S3 Source connector reads data
exported to S3 by the [Amazon S3 Sink connector](https://docs.confluent.io/kafka-connect-s3-sink/current/) and publishes it
back to an Apache Kafka® topic. Depending on the format and partitioner used to write
the data to S3, this connector can write to the destination topic using the same
partitions as the original messages exported to S3 and maintain the same message
order. The connector selects folders based on the partitioner configuration and
reads each folders S3 objects in alphabetical order. Each record is read based
on the format selected. Configuration is setup to mirror the Kafka [Amazon S3
Sink connector](https://docs.confluent.io/kafka-connect-s3-sink/current/) and
should be possible to make only minor changes to the original sink
configuration.

#### IMPORTANT
The recommended practice is to create topics manually in the destination Kafka
cluster with the correct number of partitions before running the source
connector. If the topics do not exist, Connect relies on
[auto topic
creation](/kafka-connectors/self-managed/userguide.html#connect-source-auto-topic-creation) and the number of
partitions are based upon the Kafka broker defaults. If there are more
partitions in the destination cluster, the extra partitions are not used. If
there are fewer partitions in the destination cluster, the connector task
continuously checks for the presence of the required partition and
does not produce records but remains in the running state.

Be aware of the following connector actions:

* The connector ignores any S3 object with a name that does not start with the configured topics directory. This name is “/topics/” by default.
* The connector ignores any S3 object that is below the topics directory but has an extension that does not match the configured format. For example, a JSON file is ignored when `format.class` is set for Avro files.
* The connector stops and fails if the S3 object’s name does not match the expected format or is in an unexpected location.
* The number of tasks the connector can spawn is bound by the number of source partitions and the `tasks.max` value. If there are fewer source partitions than the `tasks.max` value, the connector stops spawning tasks after reaching the number of source partitions, even if the value of `tasks.mask` is greater.

Avoid the following configuration issues:

* A file with the correct extension and a valid name format e.g. `<topic>+<partition>+<offset>.<extension>`, placed in a folder of a different topic will be read normally and written to whatever topic as defined by its filename.
* If a field partitioner is incorrectly configured to match the expected folder, it can break the ordering guarantees of S3 sink that used a deterministic sink partitioner.

<a id="s3-source-connector-features"></a>

## Features

The Backup and Restore S3 Source connector includes the following features:

- [At least once delivery](#s3-source-at-least-once-delivery)
- [Multiple tasks](#s3-source-multiple-tasks)
- [Pluggable data format with or without schema](#s3-source-pluggable-data-format)
- [Non-AWS object storage support](#s3-source-object-storage-support)
- [Matching source partitioning](#s3-source-matching-partitioning)
- [Source partition ordering](#s3-source-source-partition-ordering)
- [Pluggable partitioner](#s3-source-pluggable-partitioner)

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

### At least once delivery

In the event of a task failure the connector guarantees no messages are lost,
although the last few messages may be processed again.

<a id="s3-source-multiple-tasks"></a>

### Multiple tasks

The Backup and Restore Amazon S3 Source connector supports running one or more
tasks. You can specify the number of tasks in the `tasks.max` configuration
parameter. This can lead to huge performance gains when multiple files need to
be parsed.

<a id="s3-source-pluggable-data-format"></a>

### Pluggable data format with or without schema

Out of the box, the connector supports reading data from S3 in Avro and JSON
format. Besides records with schema, the connector supports importing plain JSON
records without schema in text files, one record per line. In general, the
connector may accept any format that provides an implementation of the Format
interface.

<a id="s3-source-object-storage-support"></a>

### Non-AWS object storage support

Amazon S3 is an industry-standard object storage service. You can use the
Kafka Connect Backup and Restore S3 connector to connect object storage
storage on non-AWS cloud platforms by using a different store URL to point at
this alternative cloud platform.

<a id="s3-source-matching-partitioning"></a>

### Matching source partitioning

Messages will be put back on to the same Kafka partition for that topic when it
was written.

<a id="s3-source-source-partition-ordering"></a>

### Source partition ordering

The connector will read records back in time order in each topic-source
partition if the `DefaultPartitioner` or a `TimeBasedPartitioner` is used.
If a FieldPartitioner is used it isn’t possible to guarantee the order of these
messages.

<a id="s3-source-pluggable-partitioner"></a>

### Pluggable partitioner

The connector comes out of the box with partitioners that support default
partitioning based on Kafka partitions, field partitioning, and time-based
partitioning in days or hours. You may implement your own partitioners by
extending the Partitioner class. Additionally, you can customize time based
partitioning by extending the TimeBasedPartitioner class.

## Limitations

The Backup and Restore Amazon S3 Source connector has the following limitations:

- The connector does not read (or restore) keys and headers exported by the
  Amazon S3 Sink connector.
- The connector does not support the `s3.proxy.url` configuration property.
- Partitioners recognize new topic folders based on polling intervals and
  task configuration updates. The `DefaultPartitioner` detects new
  partition folders. The `FieldPartitioner` notices new folders for the
  fields specified using `partition.field.name`. However, the
  `TimeBasedPartitioner` does not detect new files for a new time period.

  It is important to note that for the `TimeBasedPartitioner`, the capacity to
  scale the connector across various time ranges is limited in Backup and
  Restore mode. Currently, the connector does not support processing data that
  spans several years.
- A continuous feedback loop is created if both the [Amazon S3 Sink connector](https://docs.confluent.io/kafka-connect-s3-sink/current/overview.html) and
  the source connector are using the same Kafka cluster. This means the source
  connector is writing to the same topic being consumed by the sink connector.
  The resulting continuous feedback loop creates an ever-increasing number of
  duplicate Kafka records and S3 objects. It is possible to avoid the feedback
  loop by writing to a different topic than the one being consumed by the sink
  connector. You can use the [RegexRouter
  SMT](/platform/current/connect/transforms/regexrouter.html) with the source connector to change
  the names of the topics where the records are written. Or, you can use the
  [ExtractTopic SMT](/platform/current/connect/transforms/extracttopic.html) with the
  source connector to change the topic name based on a message field name.

  Note that the S3 Source connector fails when attempting to use the Filter SMT.
  This is due to an issue with the Connect framework (`ce-kafka`) where the
  connector stops sourcing further files if all records in a file are filtered
  through SMT.
- The S3 Sink connector can compress files before pushing them to the S3 bucket.
  However, the S3 Source connector cannot read compressed files from the S3
  bucket. Be sure to leave files uncompressed if you are storing data that will
  later be pulled from storage by the S3 source connector.
- Source data written to Confluent Platform and Confluent Cloud by the S3 Source connector must be in
  the original format written by the S3 Sink connector.

<a id="s3source-connector-install"></a>

## Install the Backup and Restore Amazon S3 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.
- An installation of the latest (`latest`) connector version.

  To install the `latest` connector version, navigate to your Confluent Platform
  installation directory and run the following command:
  ```bash
  confluent connect plugin install confluentinc/kafka-connect-s3-source: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-s3-source:1.0.0-preview
  ```

You can install a specific version by replacing `latest` with a version number. For example:

### Install the connector manually

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

<a id="s3source-connector-license"></a>

## 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.

See [Confluent license properties](https://docs.confluent.io/kafka-connect-s3-source/current/configuration_options.html#confluent-license-properties) for license properties and information about the license topic.

## Configuration Properties

For a complete list of configuration properties for this connector, see
[Configuration Reference for Amazon S3 Source Connector for Confluent Platform](../configuration_options.md#s3-source-configuration-options).

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).

<a id="s3-source-migration"></a>

## Upgrading to Version 3.x

Starting with version 3.x, the Backup and Restore Amazon S3 Source connector uses the AWS SDK v2, upgrading from v1.
This upgrade does not allow backward compatibility with versions 10.x. If you use any custom or
built-in AWS credentials providers, you must update your implementation to ensure compatibility
with the new AWS SDK v2.

Follow the steps below based on the type of credentials provider your implementation uses.

### Custom credentials provider

- Update [credentials provider](#s3-source-credentials-providers) interface: If you have implemented a
  custom credentials provider, you must update your code to implement the new AWS SDK v2 interface.

  Old interface: [com.amazon.auth.AWSCredentialsProvider](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/com/amazonaws/auth/AWSCredentialsProvider.html?session_ref=direct)

  New interface: [software.amazon.awssdk.auth.credentials.AwsCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html)
- Set a new instantiation method: The connector no longer uses a default no-args constructor to create
  an instance of your custom provider. Instead, your credentials provider class must now implement
  a static, no-argument `create()` method that returns a new instance of your provider.

### AWS built-in credentials provider

- Use v2 equivalent: If you use a credentials provider built into the AWS SDK itself, you must ensure you use the v2
  equivalent of that provider.
- Use `create()` method: This provider must implement a static, no-argument `create()`
  method that returns a new provider instance.

### Default credentials provider

- Review changes to the [default chain](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/credentials-chain.html):
  The default credentials provider chain in the AWS SDK v2 has changed the order in which
  it searches for credentials. The SDK v2 version now checks for system properties before checking
  for environment variables. This change may affect your connector’s authentication if you
  previously relied on environment variables and also have system properties set.

For more information about credentials provider migration, see
[Credentials provider changes for AWS SDK v2](https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html).

<a id="s3-source-credentials-providers"></a>

## AWS Credentials

The following sections provide information about how to configure an S3
connector to provide credentials when connecting to AWS.

### Credentials provider chain

By default, the S3 connector looks for S3 credentials in the following locations and in the following order:

1. The `aws.accessKeyId` and `aws.secretAccessKey` Java system properties on the Connect worker processes where the connector will be deployed. However, these variables are only recognized by the AWS SDK for Java and are not recommended.
2. The `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` environment variables accessible to the Connect worker processes where the connector will be deployed. These variables are recognized by the AWS CLI and all AWS SDKs (except for the AWS SDK for .NET). You use export to set these variables.
   ```bash
   export AWS_ACCESS_KEY_ID=<your_access_key_id>
   export AWS_SECRET_ACCESS_KEY=<your_secret_access_key>
   ```
3. A call will be made to the AWS Security Token Service (AWS STS) using the configured environment variables or system properties to fetch the credentials. The credentials provider looks for the following environment variables or JVM system properties:
   `AWS_WEB_IDENTITY_TOKEN_FILE` or `aws.webIdentityTokenFile`, `AWS_ROLE_ARN` or `aws.roleArn` and `AWS_ROLE_SESSION_NAME` or `aws.roleSessionName` (optional).
4. The `~/.aws/credentials` file located in the home directory of the operating system user that runs the Connect worker processes. These credentials are recognized by most AWS SDKs and the AWS CLI. Use the following AWS CLI command to create the credentials file:
   ```bash
   aws configure
   ```

   You can also manually create the credentials file using a text editor. The file should contain lines in the format shown in the example below. See [AWS Credentials File Format](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-file-format) for additional details.
   ```bash
   [default]
   aws_access_key_id = <your_access_key_id>
   aws_secret_access_key = <your_secret_access_key>
   ```

   #### NOTE
   When creating the credentials file, make sure that the user creating the credentials file is the same user that runs the Connect worker processes and that the credentials file is in this user’s home directory. Otherwise, the S3 connector will not be able to find the credentials.
5. A query sent to `http://169.254.170.2${AWS_CONTAINER_CREDENTIALS_RELATIVE_URI}` to return AWS credentials. This is applicable only if the Connect worker processes are running in AWS containers.
6. A metadata query that returns credentials from an EC2 instance. This is applicable only if the Connect worker processes are running in EC2 instances.

Choose one of the above to define the AWS credentials that the S3 connectors use, verify the credentials implementation is set correctly, and then restart all of the Connect worker processes.

#### NOTE
Confluent recommends using either **Environment variables** or a **Credentials file** because these are the most straightforward, and they can be checked using the AWS CLI tool before running the connector.

All S3 connectors run in a single Connect worker cluster and use the same credentials. This is sufficient for many use cases. If you want more control, refer to the following section to learn more about controlling and customizing how the S3 connector gets AWS credentials.

### Credentials providers

A credentials provider is a Java class that implements the
[software.amazon.awssdk.auth.credentials.AwsCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html)
interface in the AWS Java library and returns AWS credentials from the
environment. By default the S3 connector configuration property
`s3.credentials.provider.class` uses the
[software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/DefaultCredentialsProvider.html)
class. This class and interface implementation chains together six other
credential provider classes. The [DefaultCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/DefaultCredentialsProvider.html)
implementation looks for credentials in the following order:

1. **Java system properties** using the [SystemPropertyCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/SystemPropertyCredentialsProvider.html) class implementation. This implementation uses Java system properties `aws.accessKeyId` and `aws.secretAccessKey`.
2. **Environment variables** using the [EnvironmentVariableCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/EnvironmentVariableCredentialsProvider.html) class implementation. This implementation uses environment variables `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY`.
3. **WebIdentityTokenFileCredentialsProvider** using the [WebIdentityTokenFileCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/WebIdentityTokenFileCredentialsProvider.html) class implementation. This implementation uses a token file, role ARN, and role session name to return AWS credentials. The environment variables `AWS_WEB_IDENTITY_TOKEN_FILE`, `AWS_ROLE_ARN`, and `AWS_ROLE_SESSION_NAME` must be set for this provider to work. For additional information about setting up this provider, see [Using Web Identity Federation to Assume Roles for Service Accounts](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_providers_create_oidc.html).
4. **Credentials file** using the [ProfileCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/ProfileCredentialsProvider.html) class implementation. This implementation uses a credentials file located in the path `~/.aws/credentials`. This credentials provider can be used by most AWS SDKs and the AWS CLI. Use the following AWS CLI command to create the credentials file:
   ```bash
   aws configure
   ```

   You can also manually create the credentials file using a text editor. The file should contain lines in the format shown in the example below. See [AWS Credentials File Format](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html#credentials-file-format) for additional details.
   ```bash
   [default]
   aws_access_key_id = <your_access_key_id>
   aws_secret_access_key = <your_secret_access_key>
   ```

   #### NOTE
   When creating the credentials file, make sure that the user creating the credentials file is the same user that runs the Connect worker processes and that the credentials file is in this user’s home directory. Otherwise, the S3 connector will not be able to find the credentials.
5. **Amazon Elastic Container Service (ECS) container credentials** using the [ContainerCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/ContainerCredentialsProvider.html) class implementation. This implementation uses a query sent to `http://169.254.170.2${AWS_CONTAINER_CREDENTIALS_RELATIVE_URI}` to return AWS credentials for the S3 connector. For this provider to work, the environment variable `AWS_CONTAINER_CREDENTIALS_RELATIVE_URI` or `AWS_CONTAINER_CREDENTIALS_FULL_URI` must be set. For additional information about setting up this query, see [IAM Roles for Tasks](https://docs.aws.amazon.com/AmazonECS/latest/developerguide/task-iam-roles.html).
6. **EC2 instance profile credentials** using the [InstanceProfileCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/InstanceProfileCredentialsProvider.html) class implementation. EC2 instance metadata is queried for credentials. See [Amazon EC2 metadata service](https://docs.aws.amazon.com/AWSEC2/latest/UserGuide/ec2-instance-metadata.html) for additional information about instance metadata queries. For additional information and updates from AWS, see [Working with AWS credentials](https://docs.aws.amazon.com/sdk-for-java/v1/developer-guide/credentials.html).

   #### NOTE
   EC2 instance profile credentials can be used only if the environment variable `AWS_CONTAINER_CREDENTIALS_RELATIVE_URI` is not set. For more information, see [EC2ContainerCredentialsProviderWrapper](https://docs.aws.amazon.com/AWSJavaSDK/latest/javadoc/index.html?com/amazonaws/auth/EC2ContainerCredentialsProviderWrapper.html).

### Using trusted account credentials

This connector can assume a role and use credentials from a separate trusted
account. This is a default feature provided with recent versions of this
connector that include an updated version of the AWS SDK.

#### IMPORTANT
You cannot use assumed role credentials to access AWS through a proxy server without first passing [environment variables](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-proxy.html) or [system properties](https://confluence.atlassian.com/kb/how-to-configure-outbound-http-and-https-proxy-for-your-atlassian-application-834000120.html). This is due to an AWS SDK [limitation](https://github.com/aws/aws-sdk-java/issues/2558).

After you create the trust relationship, an IAM user or an application from the trusted account can
use the [AWS Security Token Service (AWS STS)](https://docs.aws.amazon.com/STS/latest/APIReference/API_AssumeRole.html)
`AssumeRole` API operation. This operation provides temporary security credentials that enable
access to AWS resources for the connector. For details, see
[Creating a Role to Delegate Permissions to an IAM User](https://docs.aws.amazon.com/IAM/latest/UserGuide/id_roles_create_for-user.html).

Example:
: ```bash
  Profile in ~/.aws/credentials:
  <br/>
  [default]
  role_arn=arn:aws:iam::037803949979:role/kinesis_cross_account_role
  source_profile=staging
  role_session_name = OPTIONAL_SESSION_NAME
  <br/>
  [staging]
  aws_access_key_id = <STAGING KEY>
  aws_secret_access_key = <STAGING SECRET>
  ```

To allow the connector to assume a role with the right permissions, set the
[Amazon Resource Name (ARN)](https://docs.aws.amazon.com/credref/latest/refdocs/setting-global-role_arn.html)
for this role. Additionally, you must choose between `source_profile` or `credential_source`
as the way to get credentials that have permission to assume the role, in the environment where the
connector is running.

#### NOTE
When setting up trusted account credentials, be aware that the approach of loading profiles from
both `~/.aws/credentials` and `~/.aws/config` does not work when configuring this connector.
Assumed role settings and credentials must be placed in the `~/.aws/credentials` file.

Additionally, the connector implements the `AwsAssumeRoleCredentialsProvider` which means
you can use the following configs to configure the assume role operation.

```bash
s3.credentials.provider.class=io.confluent.connect.s3.auth.AwsAssumeRoleCredentialsProvider
s3.credentials.provider.sts.role.arn=arn:aws:iam::012345678901:role/my-restricted-role
s3.credentials.provider.sts.role.session.name=session-name
s3.credentials.provider.sts.role.external.id=external-id
```

### Using Other Implementations

You can use a different credentials provider. To do this, set the `s3.credentials.provider.class` property to the name of any class that implements the [software.amazon.awssdk.auth.credentials.AwsCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html) interface.

Complete the following steps to use a different credentials provider:

1. Find or create a Java credentials provider class that implements the [software.amazon.awssdk.auth.credentials.AwsCredentialsProvider](https://sdk.amazonaws.com/java/api/latest/software/amazon/awssdk/auth/credentials/AwsCredentialsProvider.html) interface.
2. Put the class file in a JAR file.
3. Place the JAR file in the `share/java/kafka-connect-s3` directory on **all Connect workers**.
4. Restart the Connect workers.
5. Change the S3 connector property file to use your custom credentials. Add the provider class entry `s3.credentials.provider.class=<className>` in the S3 connector properties file.

   #### IMPORTANT
   You must use the fully qualified class name in the `<className>` entry.

<a id="s3source-connector-quickstart"></a>

## Quick Start

The following uses the `S3SinkConnector` to write a file from the Kafka topic named `s3_topic` to S3.
Then, the `S3SourceConnector` loads that Avro file from S3 to the Kafka topic named `copy_of_s3_topic`.

1. Follow the instructions from [the S3 Sink connector quick start](https://docs.confluent.io/kafka-connect-s3-sink/current/overview.html#quick-start) to set up the data to use below.
2. Install the connector through the [Confluent Hub Client](https://docs.confluent.io/current/connect/managing/confluent-hub/client.html).
   ```bash
   # run from your Confluent Platform installation directory
   confluent connect plugin install confluentinc/kafka-connect-s3-source:latest
   ```
3. Create a `quickstart-s3source.properties` file with the following contents or use the `quickstart-s3source.properties`.:
   ```properties
   name=s3-source
   tasks.max=1
   connector.class=io.confluent.connect.s3.source.S3SourceConnector
   s3.bucket.name=confluent-kafka-connect-s3-testing
   format.class=io.confluent.connect.s3.format.avro.AvroFormat
   confluent.license=
   confluent.topic.bootstrap.servers=localhost:9092
   confluent.topic.replication.factor=1
   ```
4. Edit the `quickstart-s3source.properties` to add the following properties:
   ```properties
   transforms=AddPrefix
   transforms.AddPrefix.type=org.apache.kafka.connect.transforms.RegexRouter
   transforms.AddPrefix.regex=.*
   transforms.AddPrefix.replacement=copy_of_$0
   ```

   #### IMPORTANT
   Adding this renames the output of topic of the messages to `copy_of_s3_topic`. This prevents a continuous feedback loop of messages.
5. Load the Backup and Restore S3 Source connector.
   ```bash
   confluent local load s3-source --config quickstart-s3source.properties
   ```

   #### IMPORTANT
   Don’t use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/index.html) in production environments.
6. Confirm that the connector is in a `RUNNING` state.
   ```bash
   confluent local status s3-source
   ```
7. 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 copy_of_s3_topic \
       --from-beginning | jq '.'
   ```
8. The response should be 18 records as follows.
   ```bash
   {"f1": "value1"}
   {"f1": "value2"}
   {"f1": "value3"}
   {"f1": "value4"}
   {"f1": "value5"}
   {"f1": "value6"}
   {"f1": "value7"}
   {"f1": "value8"}
   {"f1": "value9"}
   {"f1": "value1"}
   {"f1": "value2"}
   {"f1": "value3"}
   {"f1": "value4"}
   {"f1": "value5"}
   {"f1": "value6"}
   {"f1": "value7"}
   {"f1": "value8"}
   {"f1": "value9"}
   ```

## Troubleshooting Connector and Task Failures

### Stack Trace

You can use the Connect [Kafka Connect REST Interface](/platform/current/connect/references/restapi.html) to check the
status of the connectors and tasks. If a task or connector has failed, the
`trace` field will include a reason and a stack trace.

### Stack Trace message: No new files ready after scan task…

The Kafka Connect Amazon S3 Source connector only reads data exported to an S3
bucket by the Kafka Connect Amazon S3 Sink connector. The connector always
ignores a file which is not in `<topic>+<partition>+<offset>.<extension>`
format. This message occurs when the files in the S3 directory do not match the
pattern of `<topic><partition><offset>.<extension>` format expected by the
source connector.

#### Error Handling

The following `behavior.on.error` configuration properties set how the connector handles errors.

* `fail` : The connector stops processing when an error occurs. The full batch of records will not be sent to Kafka if any record in the batch is corrupted.
* `ignore` : The corrupted record is ignored. The connector continues processing the next record. For Avro, the connector ignores the file containing a corrupted record and continues processing records for the next file.
* `log` : Logs an error message and continues processing the next record. For Avro, the connector ignores the file containing a corrupted record and continues processing records for the next file.

#### NOTE
The connector always ignores a file which is not in `<topic>+<partition>+<offset>.<extension>` format.