Amazon Kinesis Source Connector for Confluent Platform¶
The Kafka Connect Kinesis Source connector is used to pull data from Amazon Kinesis and persist the data to an Apache Kafka® topic.
Features¶
At least once delivery¶
This connector 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.
Install the Kinesis Connector¶
You can install this connector by using the Confluent Hub client installation instructions or by manually downloading the ZIP file.
Prerequisites¶
Note
You must install the connector on every machine where Connect will run.
An installation of the Confluent Hub Client.
Note
This is installed by default with Confluent Enterprise.
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:confluent-hub install confluentinc/kafka-connect-kinesis:latest
You can install a specific version by replacing
latest
with a version number as shown in the following example:confluent-hub install confluentinc/kafka-connect-kinesis:1.1.4
Install the connector manually¶
Download and extract the ZIP file for your connector and then follow the manual connector installation instructions.
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 keys to subscribers, along with enterprise-level support for Confluent Platform and your connectors. If you are a subscriber, you can contact Confluent Support at support@confluent.io for more information.
See Confluent Platform license for license properties and License topic configuration for information about the license topic.
Configuration Properties¶
For a complete list of configuration properties for this connector, see Amazon Kinesis Source Connector Configuration Properties.
Note
For an example of how to get Kafka Connect connected to Confluent Cloud, see Distributed Cluster.
Usage Notes¶
The default credentials provider is DefaultAWSCredentialsProviderChain
. For
more information, see the AWS documentation.
Examples¶
Streaming ETL Demo¶
Streaming ETL Demo¶
To evaluate the Kafka Connect Kinesis Source Connector, AWS S3 Sink Connector, Azure Blob Sink Connector, and GCP GCS Sink Connector in an end-to-end streaming deployment, refer to the Cloud ETL demo. This demo also allows you to evaluate the real-time data processing capabilities of ksqlDB.

Property-based example¶
This configuration is used typically along with standalone workers.
name=KinesisSourceConnector1
connector.class=io.confluent.connect.kinesis.KinesisSourceConnector
tasks.max=1
aws.access.key.id=< Optional Configuration >
aws.secret.key.id=< Optional Configuration >
kafka.topic=< Required Configuration >
kinesis.stream=< Required Configuration >
kinesis.region=< Optional Configuration - defaults to US_EAST_1 >
confluent.topic.bootstrap.servers=localhost:9092
confluent.topic.replication.factor=1
REST-based example¶
This configuration is used typically along with distributed
workers. Write the following JSON to
connector.json
, configure all of the required values, and use the command
below to post the configuration to one the distributed connect worker(s). For
more information, see Kafka Connect REST
API
Connect distributed REST-based example:
{
"config" : {
"name" : "KinesisSourceConnector1",
"connector.class" : "io.confluent.connect.kinesis.KinesisSourceConnector",
"tasks.max" : "1",
"aws.access.key.id" : "< Optional Configuration >",
"aws.secret.key.id" : "< Optional Configuration >",
"kafka.topic" : "< Required Configuration >",
"kinesis.stream" : "< Required Configuration >"
}
}
Use curl to post the configuration to one of the Kafka Connect Workers. Change http://localhost:8083/ the endpoint of one of your Kafka Connect worker(s).
Create a new connector:
curl -s -X POST -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors
Update an existing connector:
curl -s -X PUT -H 'Content-Type: application/json' --data @connector.json http://localhost:8083/connectors/KinesisSourceConnector1/config
Quick Start¶
The Kinesis connector is used to import data from Kinesis streams, and write them into a Kafka topic. Before you begin, create a Kinesis stream and have a user profile with read access to it.
Preliminary Setup¶
Navigate to your Confluent Platform installation directory and run this command to install the latest connector version.
confluent-hub install confluentinc/kafka-connect-kinesis:latest
You can install a specific version by replacing latest with a version number. For example:
confluent-hub install confluentinc/kafka-connect-kinesis:1.1.1-preview
Adding a new connector plugin requires restarting Connect. Use the Confluent CLI to restart Connect.
Tip
The command syntax for the Confluent CLI development commands changed in 5.3.0.
These commands have been moved to confluent local
. For example, the syntax for confluent start
is now
confluent local services start
. For more information, see confluent local.
confluent local services connect stop && confluent local services connect start
Your output should resemble:
Using CONFLUENT_CURRENT: /Users/username/Sandbox/confluent-snapshots/var/confluent.NuZHxXfq
Starting Zookeeper
Zookeeper is [UP]
Starting Kafka
Kafka is [UP]
Starting Schema Registry
Schema Registry is [UP]
Starting Kafka REST
Kafka REST is [UP]
Starting Connect
Connect is [UP]
Check if the Kinesis plugin has been installed correctly and picked up by the plugin loader:
curl -sS localhost:8083/connector-plugins | jq .[].class | grep kinesis
"io.confluent.connect.kinesis.KinesisSourceConnector"
Kinesis Setup¶
You can use the AWS Management Console to set up your Kinesis stream, or by completing the following steps:
Sign up for an AWS account.
Set up your AWS credentials.
Important
Be sure to set the following permissions in your AWS Kinesis policy:
"Action": [ "kinesis:DescribeStream", "kinesis:GetShardIterator", "kinesis:GetRecords" ]
-
aws kinesis create-stream --stream-name my_kinesis_stream --shard-count 1
Insert records into your stream.
aws kinesis put-record --stream-name my_kinesis_stream --partition-key 123 --data test-message-1
The previous example shows that a record containing partition key 123
and
text test-message-1
is inserted into a stream, my_kinesis_stream
.
Source Connector Configuration¶
Start the services using the Confluent CLI:
confluent local services start
Create a configuration file named kinesis-source-config.json
with the following
contents.
{
"name": "kinesis-source",
"config": {
"connector.class": "io.confluent.connect.kinesis.KinesisSourceConnector",
"tasks.max": "1",
"kafka.topic": "kinesis_topic",
"kinesis.region": "US_WEST_1",
"kinesis.stream": "my_kinesis_stream",
"confluent.license": "",
"name": "kinesis-source",
"confluent.topic.bootstrap.servers": "localhost:9092",
"confluent.topic.replication.factor": "1"
}
}
The important configuration parameters used here are:
- kinesis.stream.name: The Kinesis Stream to subscribe to.
- kafka.topic: The Kafka topic in which the messages received from Kinesis are produced.
- tasks.max: The maximum number of tasks that should be created for this connector. Each Kinesis shard is allocated to a single task. If the number of shards specified exceeds the number of tasks, the connector throws an exception and fails.
- kinesis.region: The region where the stream exists. Defaults to
US_EAST_1
if not specified. - You may pass your AWS credentials to the Kinesis connector through your source connector configuration. To pass AWS credentials in the source configuration set the aws.access.key.id and the aws.secret.key.id parameters.
"aws.acess.key.id":<your-access-key>
"aws.secret.key.id":<your-secret-key>
Run this command to start the Kinesis source connector.
Caution
You must include a double dash (
--
) between the topic name and your flag. For more information, see this post.
confluent local services connect connector load source-kinesis --config source-kinesis-config.json
To check that the connector started successfully view the Connect worker’s log by running:
confluent local services connect log
Start a Kafka Consumer in a separate terminal session to view the data exported by the connector into the kafka topic
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic kinesis_topic --from-beginning
Finally, stop the Confluent services using the command:
confluent local stop
Remove unused resources¶
Delete your stream and clean up resources to avoid incurring any unintended charges.
aws kinesis delete-stream --stream-name my_kinesis_stream
AWS Credentials¶
By default, the kinesis connector looks for kinesis credentials in the following locations and in the following order:
- The
AWS_ACCESS_KEY_ID
andAWS_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.
export AWS_ACCESS_KEY_ID=<your_access_key_id>
export AWS_SECRET_ACCESS_KEY=<your_secret_access_key>
The ``AWS_ACCESS_KEY`` and ``AWS_SECRET_KEY`` can be used instead, but are not recognized by the AWS CLI.
The
aws.accessKeyId
andaws.secretKey
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.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:aws configure
You can also manually create the credentials file using a text editor. The file should contain lines in the following format:
[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 kinesis connector will not be able to find the credentials.
See AWS Credentials File Format for additional details.
Choose one of the above to define the AWS credentials that the kinesis 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 kinesis 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 kinesis connector gets AWS credentials.
Caution
If you configure one of the AWS key and AWS secret key implementations (as detailed above), credentials can not also be supplied through the following Credentials Providers or by using the Trusted Account Credentials implementation. Attempting to provide credentials using multiple implementations will cause authentication failure.
Credentials Providers¶
A credentials provider is a Java class that implements the com.amazon.auth.AWSCredentialsProvider interface in the AWS Java library and returns AWS credentials from the environment. By default the kinesis connector configuration property kinesis.credentials.provider.class
uses the com.amazon.auth.DefaultAWSCredentialsProviderChain class. This class and interface implementation chains together five other credential provider classes.
The com.amazonaws.auth.DefaultAWSCredentialsProviderChain implementation looks for credentials in the following order:
- Environment variables using the com.amazonaws.auth.EnvironmentVariableCredentialsProvider class implementation. This implementation uses environment variables
AWS_ACCESS_KEY_ID
andAWS_SECRET_ACCESS_KEY
. Environment variablesAWS_ACCESS_KEY
andAWS_SECRET_KEY
are also supported by this implementation; however, these two variables are only recognized by the AWS SDK for Java and are not recommended. - Java system properties using the com.amazonaws.auth.SystemPropertiesCredentialsProvider class implementation. This implementation uses Java system properties
aws.accessKeyId
andaws.secretKey
. - Credentials file using the com.amazonaws.auth.profile.ProfileCredentialsProvider 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:
aws configure
You can also manually create the credentials file using a text editor. The file should contain lines in the following format:
[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 Kinesis connector will not be able to find the credentials.
See AWS Credentials File Format for additional details.
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 or system properties. This is due to an AWS SDK limitation.
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)
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.
- Example:
Profile in ~/.aws/credentials: [default] role_arn=arn:aws:iam::037803949979:role/kinesis_cross_account_role source_profile=staging role_session_name = OPTIONAL_SESSION_NAME [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)
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.
Using Other Implementations¶
You can use a different credentials provider. To do this, set the
kinesis.credentials.provider.class
property to the name of any class that
implements the com.amazon.auth.AWSCredentialsProvider
interface.
Important
If you are using a different credentials provider, do not include the aws.access.key.id
and aws.secret.key.id
in the connector configuration file. If these parameters are included, they will override the custom credentials provider class.
Complete the following steps to use a different credentials provider:
Find or create a Java credentials provider class that implements the com.amazon.auth.AWSCredentialsProvider interface.
Put the class file in a JAR file.
Place the JAR file in the
share/java/kafka-connect-kinesis
directory on all Connect workers.Restart the Connect workers.
Change the kinesis connector property file to use your custom credentials. Add the provider class entry
kinesis.credentials.provider.class=<className>
in the kinesis connector properties file.Important
You must use the fully qualified class name in the
<className>
entry.