Configuring Kafka Connect
1. Setting Bootstrap Servers
| Property | Default Value | Description |
|---|---|---|
| bootstrap.servers | localhost:9092 | Comma-separated list of host/port pairs to use for establishing the initial connection to the Kafka cluster. |
Example: Configuring Bootstrap Servers
{
"bootstrap.servers": "kafka1:9092,kafka2:9092,kafka3:9092"
}Note: Ensure all Kafka brokers in the cluster are reachable from the Connect worker.
2. Configuring Key and Value Converters
| Converter | Description | Default | Notes |
|---|---|---|---|
| key.converter | Converter for record keys | org.apache.kafka.connect.json.JsonConverter |
Used to serialize/deserialize keys in Kafka messages |
| value.converter | Converter for record values | org.apache.kafka.connect.json.JsonConverter |
Used to serialize/deserialize values in Kafka messages |
Example: Configuring JSON Converters
{
"key.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"key.converter.schemas.enable": "true",
"value.converter.schemas.enable": "true"
}Example: Configuring Avro Converters
{
"key.converter": "io.confluent.connect.avro.AvroConverter",
"value.converter": "io.confluent.connect.avro.AvroConverter",
"key.converter.schema.registry.url": "http://localhost:8081",
"value.converter.schema.registry.url": "http://localhost:8081"
}
Note: When using Avro converters, ensure that the schema registry is accessible and properly configured.
Key vs Value Converter Differences
- Key Converters: Handle serialization/deserialization of Kafka message keys
- Value Converters: Handle serialization/deserialization of Kafka message values
3. Setting Offset Storage
| Property | Default Value | Description |
|---|---|---|
| offset.storage.topic | connect-offsets |
Topic name used for storing connector offsets |
| offset.storage.replication.factor | 3 |
Replication factor for the offsets topic |
| offset.storage.partitions | 25 |
Number of partitions for the offsets topic |
| offset.storage.commit.policy | IntervalCommitPolicy |
Policy for committing offsets (e.g., IntervalCommitPolicy, AlwaysKafkaCommitPolicy) |
Example: Configure Offset Storage
{
"offset.storage.topic": "my-connect-offsets",
"offset.storage.replication.factor": 3,
"offset.storage.partitions": 25
}4. Configuring Config Storage
| Property | Default Value | Description |
|---|---|---|
| config.storage.topic | kafka-connect-configs |
Topic name used for storing connector configurations |
| config.storage.replication.factor | 3 |
Replication factor for the config storage topic |
| config.storage.partitions | 12 |
Number of partitions for the config storage topic |
| config.storage.class | org.apache.kafka.connect.storage.FileConfigStorage |
Class used to store configurations (e.g., FileConfigStorage, KafkaConfigStorage) |
Example: Configuring Kafka-based Storage
{
"config.storage.topic": "connect-configs",
"config.storage.replication.factor": 3,
"config.storage.partitions": 12,
"offset.storage.topic": "connect-offsets",
"status.storage.topic": "connect-status"
}
Note: When using Kafka-based config storage, ensure the topic is created with appropriate replication and partition settings to support high availability.
5. Setting Status Storage
| Property | Default Value | Description |
|---|---|---|
| status.storage.topic | _kafka_connect_status |
Topic name used for storing connector and task statuses. |
| status.storage.partitions | 1 |
Number of partitions for the status storage topic. |
| status.storage.replication.factor | 3 |
Replication factor for the status storage topic. |
| status.storage.min.insync.replicas | 2 |
Minimum number of in-sync replicas required for status topic. |
Example: Configure Status Storage Topic
{
"status.storage.topic": "_connect_status",
"status.storage.partitions": 5,
"status.storage.replication.factor": 3,
"status.storage.min.insync.replicas": 2
}6. Configuring Replication Factor
| Property | Default Value | Description |
|---|---|---|
| replication.factor | 1 |
Controls the number of replicas for each partition in a topic. Higher values increase fault tolerance but also storage overhead. |
| min.insync.replicas | 1 |
Minimum number of replicas that must acknowledge a write before it is considered successful. Ensures data durability. |
Example: Setting Replication Factor for a Topic
kafka-topics.sh --create --topic my-topic --partitions 3 --replication-factor 3 --bootstrap-server localhost:9092
Note: The replication factor must be less than or equal to the number of brokers in the cluster.
7. Setting Group ID
| Property | Default Value | Description |
|---|---|---|
| group.id | connect-cluster |
The unique identifier for the Kafka Connect cluster. This ID is used to group connectors and tasks together. |
Note: All workers in the same Kafka Connect cluster must have the same
group.id to ensure proper coordination and task distribution.
8. Configuring Plugin Path
| Property | Default | Description |
|---|---|---|
| plugin.path | none |
Comma-separated list of paths to search for Connect plugins. These paths must contain plugin JARs. |
Note: Plugin paths must be absolute or relative to the Kafka installation directory.
Warning: Ensure that all plugin JARs are compatible with the Kafka Connect version in use.
9. Setting REST Port
| Property | Default Value | Description |
|---|---|---|
| rest.port | 8083 |
The port on which the Kafka Connect REST API listens. |
10. Configuring Internal Topics
| Property | Default Value | Description |
|---|---|---|
| internal.topic.replication.factor | 3 |
Replication factor for internal topics (e.g., connect-configs, connect-status) |
| internal.topic.partitions | 1 |
Number of partitions for internal topics |
| config.storage.topic | connect-configs |
Name of the topic used to store connector configurations |
| status.storage.topic | connect-status |
Name of the topic used to store connector status |
| offset.storage.topic | connect-offsets |
Name of the topic used to store connector offsets |
Example: Internal Topic Configuration
{
"config.storage.topic": "my-connect-configs",
"status.storage.topic": "my-connect-status",
"offset.storage.topic": "my-connect-offsets",
"internal.topic.replication.factor": 3,
"internal.topic.partitions": 1
}
Note: Internal topics are automatically created by Kafka Connect if they don't exist, but it's recommended to configure them explicitly for production environments.