A Kafka Connect connector integrates Kafka with an external system, moving data into Kafka or out to databases, search engines, object stores, warehouses, and other applications. It provides a production-oriented alternative to writing separate producer and consumer integrations for every system. section 5.1 section 5.7
A connector coordinates the work; tasks perform the actual data movement. Source connectors bring records from an external system into Kafka, while sink connectors deliver Kafka records externally. The connector supplies each task with configuration, including the settings needed to divide work across the permitted number of tasks. section 5.6.1
Choose and configure a connector at two levels:
- Configure the Connect worker with defaults such as key and value converters, plugin locations, and—when applicable—a source-connector consumer group ID. section 5.3
- Configure the individual connector with its implementation class, external connection details, data-selection settings, converters, and tasks.max. section 5.3
- Run Connect standalone for local development or distributed across instances for parallelism and fault tolerance. section 5.3
- Prefer an existing connector when it fits; create one when available connectors and Single Message Transforms cannot meet the integration need. section 5.7
Listing 5.1 REST API call to start a connector
curl -i -X PUT http://localhost:8083/connectors/
jdbc_source_connector_example/config \
-H "Content-Type: application/json" \
-d '{
"connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
"connection.url": "jdbc:postgresql://postgres:5432/postgres",
"connection.user": "postgres",
"connection.password": "postgres",
"mode":"timestamp",
"timestamp.column.name":"ts",
"topic.prefix":"postgres_",
"value.converter":"org.apache.kafka.connect.json.JsonConverter", #1
"value.converter.schemas.enable": "false", #2
"tasks.max":"1" #3
}'
A REST request illustrates starting a JDBC source connector by supplying its class, PostgreSQL connection settings, timestamp mode, topic prefix, JSON conversion, and a one-task limit. section 5.3
Stock ticker source connector
A custom StockTickerSourceConnector extends SourceConnector and defines required settings such as an API URL and destination topic with ConfigDef. Its taskConfigs method limits task count, groups ticker symbols, and gives every task the API, topic, token, batch-size, and assigned-symbol configuration it needs. The source task then polls the HTTP API and turns returned ticker data into SourceRecord instances. section 5.6.1
Listing 5.6 Configuring the tasks
@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
List<Map<String, String>> taskConfigs = new ArrayList<>();
List<String> symbols = monitorThread.symbols(); #1
int numTasks = Math.min(symbols.size(), maxTasks);
List<List<String>> groupedSymbols =
ConnectorUtils.groupPartitions(symbols, numTasks); #2
for (List<String> symbolGroup : groupedSymbols) {
Map<String, String> taskConfig = new HashMap<>(); #3
taskConfig.put(TOPIC_CONFIG, topic);
taskConfig.put(API_URL_CONFIG, apiUrl);
taskConfig.put(TOKEN_CONFIG, token);
taskConfig.put(TASK_BATCH_SIZE_CONFIG, Integer.toString(batchSize));
taskConfig.put(TICKER_SYMBOL_CONFIG, String.join(",", symbolGroup)); #4
taskConfigs.add(taskConfig);
}
return taskConfigs;
}
The task-configuration listing shows how a connector partitions ticker symbols and creates one configuration map per task. section 5.6.1
