A SourceTask polls an external system and returns records for insertion into Kafka. Kafka Connect starts task instances from connector-generated configurations; each task runs in its own thread and may run on different workers in distributed mode. section 5.6.1 section 5.3
A SourceTask implementation typically handles:
- Polling the configured external application through poll(). section 5.6.1
- Throttling requests when the external API limits call frequency. section 5.6.1
- Waiting briefly when no records are available so the source can produce more data. section 5.6.1
- Converting returned external data into SourceRecord instances for Kafka. section 5.6.1
Stock ticker source
A StockTickerSourceTask receives a group of ticker symbols from its connector configuration, polls a stock service API, parses the JSON response, and maps each ticker result to a SourceRecord. With ten symbols and two maximum tasks, the connector can assign five symbols to each task. section 5.6.1
Listing 5.7 Start of the poll method for the source task
public List<SourceRecord> poll() throws InterruptedException {
final long nextUpdate = lastUpdate.get() + timeBetweenPoll; #1
final long now = sourceTime.milliseconds(); #2
final long sleepMs = nextUpdate - now; #3
if (sleepMs > 0) {
LOG.trace("Waiting {} ms to poll API feed next", sleepMs);
sourceTime.sleep(sleepMs);
}
The polling skeleton shows how a task calculates its next update, checks source time, and sleeps before making the next request. section 5.6.1
Listing 5.8 Core polling logic
HttpRequest request = HttpRequest.newBuilder() #1
.uri(uri)
.GET()
.headers("Content-Type", "text/plain;charset=UTF-8")
.build();
HttpResponse<String> response;
try {
response = httpClient.send(request,
HttpResponse.BodyHandlers.ofString()); #2
AtomicLong counter = new AtomicLong(0);
JsonNode apiResult = objectMapper.readTree(response.body()); #3
ArrayNode tickerResults = (ArrayNode) apiResult.get(resultNode);
LOG.debug("Retrieved {} records", tickerResults.size());
Stream<JsonNode> stockRecordStream =
StreamSupport.stream(tickerResults.spliterator(), false);
List<SourceRecord> sourceRecords = stockRecordStream.map(entry -> { #4
Map<String, String> sourcePartition =
Collections.singletonMap("API", apiUrl);
Map<String, Long> sourceOffset =
Collections.singletonMap("index", counter.getAndIncrement());
Schema schema = getValueSchema(entry); #5
Map<String, Object> resultsMap = toMap(entry); #6
return new SourceRecord(sourcePartition,
sourceOffset,
topic,
null,
schema,
toStruct(schema, resultsMap)); #7
}).toList();
lastUpdate.set(sourceTime.milliseconds()); #8
return sourceRecords; #9
}
The core polling code shows the HTTP request, JSON response parsing, and transformation of external entries into source records. section 5.6.1
