3.1 – Connector Configuration Basics

Overview

This module is the authoritative guide to configuring Kafka Connect connectors. By the end, you will understand:

• The structure of connector configurations.
• Required vs. optional properties.
• Converter configurations for key/value formats.
• How to submit, update, and manage connector configurations via REST API.

All connector configuration concepts live here. Other modules will reference this when discussing specific connectors or deployment scenarios.

Connector Configuration Structure

Configuration Format

Connectors are configured using JSON (for REST API submissions) or properties files (for standalone mode).

JSON Format (REST API)

{
  "name": "my-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "3",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "postgres-"
  }
}

Properties Format (Standalone Mode)

name=my-connector
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=3
connection.url=jdbc:postgresql://localhost:5432/mydb
mode=incrementing
incrementing.column.name=id
topic.prefix=postgres-

Key Insight: The name field is required and must be unique within a Connect cluster. All other configuration is in the config object (JSON) or as flat properties.

Core Configuration Properties

1. connector.class (Required)

• Purpose: Specifies the fully-qualified class name of the connector plugin.
• Example: io.confluent.connect.jdbc.JdbcSourceConnector
• Location: Connector JAR must be in the plugin path (plugin.path in worker config).

How to Find It: Check the connector's documentation or use:

curl http://localhost:8083/connector-plugins

2. name (Required)

• Purpose: Unique identifier for the connector within the Connect cluster.
• Constraints:
  • Must be unique.
  • Alphanumeric, hyphens, and underscores only.
  • Used in metrics, logs, and REST API endpoints.

Example: jdbc-source-users, s3-sink-logs

3. tasks.max (Required)

• Purpose: Maximum number of tasks the connector can create for parallel execution.
• Default: 1
• Impact:
  • Higher values increase throughput but also resource usage.
  • Connector plugin determines actual task count (may be less than tasks.max).

Example:

"tasks.max": "5"

Rule of Thumb: For source connectors, set tasks.max to the number of logical partitions (e.g., database tables, files). For sink connectors, set to the number of Kafka partitions being consumed.

Connector-Specific Configuration

Source Connector Example: JDBC Source

{
  "name": "jdbc-source-orders",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "connection.user": "kafka",
    "connection.password": "secret",
    "mode": "timestamp",
    "timestamp.column.name": "updated_at",
    "table.whitelist": "orders",
    "topic.prefix": "postgres-",
    "poll.interval.ms": "5000"
  }
}

Key Properties:

• connection.url, connection.user, connection.password: Database connection details.
• mode: How to detect new/updated rows (incrementing, timestamp, timestamp+incrementing, bulk).
• timestamp.column.name: Column to track for timestamp mode.
• table.whitelist: Tables to monitor (comma-separated).
• topic.prefix: Prefix for target Kafka topics (e.g., orders → postgres-orders).
• poll.interval.ms: Polling frequency (default: 5000ms).

Sink Connector Example: Elasticsearch Sink

{
  "name": "elasticsearch-sink-logs",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "tasks.max": "3",
    "topics": "logs",
    "connection.url": "http://localhost:9200",
    "type.name": "_doc",
    "key.ignore": "true",
    "schema.ignore": "true",
    "batch.size": "500",
    "max.buffered.records": "5000"
  }
}

Key Properties:

• topics: Kafka topics to consume (comma-separated).
• connection.url: Elasticsearch cluster URL.
• type.name: Elasticsearch document type (deprecated in ES 7+, use _doc).
• key.ignore: Ignore record keys (documents use Kafka offset as ID).
• batch.size: Records per bulk indexing request.
• max.buffered.records: Maximum records to buffer before blocking.

Converter Configuration

What Are Converters?

Converters control how Connect serializes/deserializes data between Kafka's byte format and Connect's internal SinkRecord/SourceRecord format.

Key Converter Types

Configuring Converters

Converters are configured separately for keys and values:

{
  "name": "jdbc-source-avro",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "postgres-",
    
    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://localhost:8081"
  }
}

Explanation:

• Key Converter: StringConverter (keys are plain strings).
• Value Converter: AvroConverter (values are Avro-encoded with Schema Registry).
• Schema Registry URL: Required for AvroConverter.

Default Converters (Worker-Level)

If converters are not specified in connector config, Connect uses worker-level defaults from connect-distributed.properties:

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

Best Practice: Always specify converters explicitly in connector config to avoid ambiguity.

Key vs. Value Formats

Independent Configuration

Keys and values can use different formats:

{
  "key.converter": "org.apache.kafka.connect.storage.StringConverter",
  "value.converter": "io.confluent.connect.avro.AvroConverter"
}

Use Case: Keys are simple strings (e.g., user ID), values are complex Avro records.

Schema Support

• With Schema (schemas.enable=true): JSON includes schema metadata.

  {
    "schema": {"type": "struct", "fields": [...]},
    "payload": {"id": 1, "name": "Alice"}
  }
  

• Without Schema (schemas.enable=false): JSON is plain payload.

  {"id": 1, "name": "Alice"}
  

Trade-off: Schema support enables better compatibility checks but increases message size.

Required vs. Optional Properties

Required Properties (All Connectors)

Connector-Specific Required Properties

Each connector defines its own required properties. For JDBC Source:

• connection.url: Database URL
• mode: Query mode (incrementing, timestamp, etc.)

How to Discover: Check connector documentation or use:

curl -X PUT http://localhost:8083/connector-plugins/<connector-class>/config/validate \
  -H "Content-Type: application/json" \
  -d '{"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector"}'

This returns a list of required and optional properties with descriptions.

Submitting Connector Configurations via REST API

Create a New Connector

Endpoint: POST /connectors

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "my-connector",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "tasks.max": "3",
      "connection.url": "jdbc:postgresql://localhost:5432/mydb",
      "mode": "incrementing",
      "incrementing.column.name": "id",
      "topic.prefix": "postgres-"
    }
  }'

Response (Success):

{
  "name": "my-connector",
  "config": {...},
  "tasks": [],
  "type": "source"
}

Response (Failure):

{
  "error_code": 400,
  "message": "Connector configuration is invalid: Missing required configuration \"connection.url\""
}

Update an Existing Connector

Endpoint: PUT /connectors/{name}/config

curl -X PUT http://localhost:8083/connectors/my-connector/config \
  -H "Content-Type: application/json" \
  -d '{
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "5",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "postgres-"
  }'

Behavior:

• Connector is restarted with the new configuration.
• Tasks are reassigned based on new tasks.max.
• Offsets are preserved (no data loss).

Validating Configurations

Validate Before Creation

Endpoint: PUT /connector-plugins/{connector-class}/config/validate

curl -X PUT http://localhost:8083/connector-plugins/io.confluent.connect.jdbc.JdbcSourceConnector/config/validate \
  -H "Content-Type: application/json" \
  -d '{
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "3",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb"
  }'

Response:

{
  "name": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "error_count": 1,
  "groups": ["Common", "Connection", "Mode"],
  "configs": [
    {
      "definition": {...},
      "value": {...},
      "recommended_values": [],
      "errors": ["Missing required configuration \"mode\""]
    }
  ]
}

Use Case: Validate configurations before submitting to avoid runtime failures.

Managing Connector Configurations

View Current Configuration

curl http://localhost:8083/connectors/my-connector/config

List All Connectors

curl http://localhost:8083/connectors

Delete a Connector

curl -X DELETE http://localhost:8083/connectors/my-connector

Note: Deleting a connector stops all tasks and removes the connector from the cluster, but does not delete Kafka topics or offsets.

Pause/Resume a Connector

Pause (stops data processing, tasks remain assigned):

curl -X PUT http://localhost:8083/connectors/my-connector/pause

Resume (restarts data processing):

curl -X PUT http://localhost:8083/connectors/my-connector/resume

Use Case: Pause connectors during maintenance windows or external system outages.

Configuration Best Practices

1. Use Descriptive Connector Names

Bad: connector1, test, temp
Good: jdbc-source-users, s3-sink-logs-prod

2. Explicitly Specify Converters

Always set key.converter and value.converter in connector config to avoid ambiguity.

3. Validate Before Deploying

Use the /config/validate endpoint to catch configuration errors early.

4. Use Environment Variables for Secrets

Avoid hardcoding credentials in config files. Use externalized secrets (covered in Module 9.1 – Security).

5. Start with Low tasks.max, Then Scale

Begin with tasks.max=1 to verify connectivity and functionality, then increase for higher throughput.

6. Document Custom Properties

If using custom SMTs or advanced configs, document them in your connector deployment scripts.

Common Configuration Errors

Error 1: "Connector class not found"

Cause: Connector plugin JAR is not in the plugin path.

Fix: Ensure connector JAR is in plugin.path directory (e.g., /usr/local/share/kafka/plugins/) and restart workers.

Error 2: "Missing required configuration"

Cause: A required property is not set.

Fix: Use /config/validate to identify missing properties.

Error 3: "Failed to connect to external system"

Cause: Incorrect connection.url, credentials, or network issues.

Fix: Test external system connectivity manually (e.g., psql, curl) before configuring connector.

Error 4: "Converter not found"

Cause: Converter class is not in the classpath or is misspelled.

Fix: Verify converter class name and ensure required dependencies (e.g., kafka-connect-avro-converter) are installed.

Converter Configuration Reference

This section provides a complete reference for all common converter configurations. Use this as a quick lookup when configuring connectors.

String Converter

key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter

Use Case: Simple string keys and values (e.g., log lines, plain text).

JSON Converter (Without Schema)

key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false

Use Case: JSON data where schema evolution is not needed.

Example Record:

{"user_id": 123, "name": "Alice"}

JSON Converter (With Embedded Schema)

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

Use Case: JSON data with inline schema for compatibility checks.

Example Record:

{
  "schema": {
    "type": "struct",
    "fields": [
      {"field": "user_id", "type": "int32"},
      {"field": "name", "type": "string"}
    ]
  },
  "payload": {"user_id": 123, "name": "Alice"}
}

Trade-off: Larger message size due to embedded schema.

JSON Converter (With Schema Registry)

key.converter=io.confluent.connect.json.JsonConverter
value.converter=io.confluent.connect.json.JsonConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081
key.converter.schemas.enable=true
value.converter.schemas.enable=true

Use Case: JSON data with centralized schema management via Schema Registry.

Benefit: Schemas are stored centrally, enabling evolution and compatibility checks.

Avro Converter (With Schema Registry)

key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081
key.converter.schemas.enable=true
value.converter.schemas.enable=true

Use Case: Production pipelines requiring schema evolution and compact binary format.

Benefits:

• Compact: Smaller message size than JSON.
• Schema Evolution: Backward/forward compatibility support.
• Type Safety: Strongly typed schema validation.

Protobuf Converter (With Schema Registry)

key.converter=io.confluent.connect.protobuf.ProtobufConverter
value.converter=io.confluent.connect.protobuf.ProtobufConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081
key.converter.schemas.enable=true
value.converter.schemas.enable=true

Use Case: Google Protocol Buffers for high-performance, strongly-typed data.

Benefits:

• Performance: Fastest serialization/deserialization.
• Cross-Language: Works with gRPC and other Protobuf-based systems.

ByteArray Converter

key.converter=org.apache.kafka.connect.converters.ByteArrayConverter
value.converter=org.apache.kafka.connect.converters.ByteArrayConverter

Use Case: Raw binary data (e.g., images, compressed files).

Troubleshooting Connector Deployment

This section provides practical troubleshooting steps for common connector deployment issues.

Step 1: Check Connector Logs

Worker Logs Location: /var/log/kafka/connect.log (or configured log path)

What to Look For:

• ERROR: Connector or task failures.
• WARN: Configuration issues, retries, slow responses.
• INFO: Startup, task assignment, offset commits.

Example Log Pattern (Error):

[2024-01-15 10:05:30,789] ERROR Failed to poll records from database (com.example.JdbcSourceTask)
org.apache.kafka.connect.errors.ConnectException: Failed to connect to database
    at com.example.JdbcSourceTask.poll(JdbcSourceTask.java:123)

Step 2: Check Connector Status via REST API

curl http://localhost:8083/connectors/my-connector/status | jq .

Example Response (Healthy):

{
  "name": "my-connector",
  "connector": {"state": "RUNNING", "worker_id": "worker1:8083"},
  "tasks": [
    {"id": 0, "state": "RUNNING", "worker_id": "worker1:8083"}
  ],
  "type": "source"
}

Example Response (Failed):

{
  "name": "my-connector",
  "connector": {"state": "FAILED", "worker_id": "worker1:8083", "trace": "..."},
  "tasks": [
    {"id": 0, "state": "FAILED", "worker_id": "worker1:8083", "trace": "SQLException: Connection refused"}
  ]
}

Action: Inspect trace field for error details.

Step 3: Verify Connection to External System

For Databases (JDBC):

# Test PostgreSQL connection
psql -h localhost -U myuser -d mydb -c "SELECT 1;"

# Test MySQL connection
mysql -h localhost -u myuser -p -e "SELECT 1;"

For REST APIs:

curl https://api.example.com/health

For S3:

aws s3 ls s3://my-bucket/ --profile myprofile

Action: Ensure network connectivity, credentials, and permissions are correct.

Step 4: Check Kafka Topic Existence and Data

List Topics:

kafka-topics --bootstrap-server localhost:9092 --list

Describe Topic:

kafka-topics --bootstrap-server localhost:9092 --describe --topic my-topic

Consume from Topic (verify data is flowing):

kafka-console-consumer --bootstrap-server localhost:9092 --topic my-topic --from-beginning --max-messages 10

Action: If topic is empty or missing, check source connector status and logs.

Step 5: Validate Schema Compatibility

For Avro/JSON Schema with Schema Registry:

# List schemas
curl http://localhost:8081/subjects

# Get schema for topic
curl http://localhost:8081/subjects/my-topic-value/versions/latest

Common Issues:

• Schema mismatch: Connector expects field user_id but data has userId.
• Type mismatch: Connector expects int32 but data has string.

Action: Use SMTs (Module 4.1) to transform field names or types, or update source/sink schema.

Step 6: Check Worker Resource Usage

CPU/Memory:

# Check worker process
top -p $(pgrep -f kafka-connect)

# Or with docker
docker stats kafka-connect

Common Issues:

• Out of Memory: Increase JVM heap size (KAFKA_HEAP_OPTS="-Xmx4G").
• High CPU: Too many tasks, reduce tasks.max or add more workers.

Troubleshooting Checklist

Use this checklist when a connector fails:

• Check connector status via REST API (/connectors/{name}/status)
• Review worker logs for ERROR/WARN messages
• Test external system connectivity (database, API, S3, etc.)
• Verify Kafka topic exists and has expected partitions
• Validate schema compatibility (if using Schema Registry)
• Check worker resource usage (CPU, memory, disk)
• Verify connector configuration (use /config/validate)
• Check network/firewall rules (if in containerized/cloud environment)

JDBC Sink: auto.create and auto.evolve

For JDBC Sink Connectors, two additional properties simplify schema management:

auto.create

{
  "auto.create": "true"
}

Purpose: Automatically creates database tables if they don't exist.

Use Case: Development/testing environments where you want rapid iteration.

Warning: Use with caution in production (may create incorrect schemas).

auto.evolve

{
  "auto.evolve": "true"
}

Purpose: Automatically evolves database schema when new fields are added to Kafka records.

Use Case: Dynamic schemas where fields are added over time.

Limitation: Only supports additive changes (new fields). Cannot remove fields or change types.

Example:

-- Initial schema
CREATE TABLE users (id INT, name VARCHAR);

-- Kafka record adds "email" field
{"id": 1, "name": "Alice", "email": "alice@example.com"}

-- auto.evolve adds column
ALTER TABLE users ADD COLUMN email VARCHAR;

Key Takeaways

1. connector.class, name, and tasks.max are required for all connectors.
2. Converters control key/value serialization; configure them explicitly per connector.
3. REST API is the primary interface for submitting and updating configurations.
4. Validation endpoint (/config/validate) catches errors before deployment.
5. Pause/Resume allows graceful connector management without deletion.
6. Best practices: Use descriptive names, validate configs, externalize secrets, start small and scale.

What's Next?

• Module 3.2 explains Standalone vs. Distributed mode deployment.
• Module 3.3 covers task parallelism and scaling strategies.
• Module 9.1 teaches secure credential management.

Further Reading

• Kafka Connect REST API Reference
• Connector Configuration Properties (Confluent)

End of Module 3.1