SourceRecord represents one record imported by a Kafka Connect source connector. It carries the value, optional schema, and source-position metadata needed to describe where the record came from. section 5.6.1
Build each SourceRecord from the external result by:
- Mapping the external location to a source partition, such as an API URL, database table, or file name. section 5.6.1
- Recording the individual result position as the source offset, allowing processing to resume from a committed position. section 5.6.1
- Converting JSON into a Map and Struct using a generated schema, which can accompany the record in Kafka. section 5.6.1
Ticker API polling
A source task polls a ticker API, parses its nested JSON array, flattens the entries, and maps each entry to a SourceRecord destined for Kafka. 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
}
Follow the polling flow from the HTTP request and JSON response toward SourceRecord creation. section 5.6.1
