A source offset identifies the position of an individual record within an external source partition. In a Kafka Connect source task, both values are supplied as Map instances when constructing a SourceRecord. section 5.6.1
The source partition describes where the connector reads data, such as a database table, file, or API URL. The source offset then identifies the record’s position within that location, allowing processing to resume from a committed position. section 5.6.1
Ticker API records
A stock-ticker connector polls an API, flattens its JSON results, and maps each entry to a SourceRecord. It can use the API URL as the source partition and the entry’s position in the retrieved result as the source offset. 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 implementation from the HTTP request and JSON response to the stream of records that receives source-partition and source-offset metadata. section 5.6.1
