A source connector brings records from an external system into Kafka topics. Kafka Connect runs it as a configurable plugin, so an available connector can be started with JSON configuration rather than application code. Typical sources include PostgreSQL or MySQL databases, MongoDB, S3, and external APIs. Source connectors complement sink connectors, which move Kafka records to external targets; several sinks can export the same topic without changing the originating application. section 5.1
The connector itself coordinates work rather than moving every record. A SourceConnector extends Connector and produces configuration for SourceTask instances. Connect starts and distributes those tasks across workers; each task runs in its own thread. The tasks.max setting is an upper bound, not a promise: the connector decides how many tasks are needed. In distributed mode, task placement across workers increases throughput and allows processing to continue when a worker fails. section 5.3 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;
}
A connector can split a monitored symbol list into groups, create one configuration map per group, and cap the number of task configurations at maxTasks. Each map carries shared API, topic, token, and batch settings plus the symbols assigned to that task. section 5.6.1
A custom connector starts by defining its required settings with ConfigDef, such as an API URL and destination topic. Defaults can be supplied for optional settings; a missing setting without a default causes ConfigException and stops the connector. The connector then turns its configuration and available work into task configurations. A source task polls the external service, parses returned data, and creates SourceRecord values for Kafka. section 5.6.1
Share one live stock feed
A fintech company can implement a source connector that consumes a costly real-time stock ticker feed once and publishes it to Kafka. Departments then consume the Kafka data in near real time instead of each purchasing a separate feed. The connector can divide ticker symbols among tasks, while each task polls the API and emits records. section 5.6 section 5.6.1
Source connectors can also place transformations at the ingestion boundary. For example, a sales-data connector can capture transactions as they occur while an SMT masks credit-card fields before records enter Kafka. Consumers can then react immediately for rewards processing or analyze the sales topic with Kafka Streams, while other connectors export derived results. section 1.5 section 5.1
