DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
will be run periodically when
offsetsare committed.does not guarantee that offsets being committed won't necessarily correspond to will necessarily be the latest offsets returned delivered by this source task’s
poll().
...
This is much needed for SourceConnector that talk to external-system that maintain some watermark (for cleanup). These connector maintain offsets in OffsetTopic + signal low-watermark to external systems. Generally, low-watermark cannot be sent out for every record, and publishing offsets at record level is more suitable. When task failsis restarted, it must start from last committed offset. So, watermark must always be lower than last-committed-offset. Thus watermark must only be updated based on latest committed offset (and not based on latest SourceRecords published SourceRecord to KafkaTopic, which can be tracked via commitRecord).
Example: Debezium PostgresSourceConnector
...
| Code Block | ||
|---|---|---|
| ||
/**
* This method is invoked periodically when offsets are committed for this source task. Note that the offsets
* being committed won't necessarily correspond to the latest offsets returned by this source task via
* {@link #poll()}. Also see {@link #commitRecord(SourceRecord, RecordMetadata)} which allows for a more
* fine-grained tracking of records that have been successfully delivered.
* <p>
* SourceTasks are not required to implement this functionality; Kafka Connect will record offsets
* automatically. This hook is provided for systems that also need to store offsets internally
* in their own system.
* <p>
* Note: Exceptions thrown by this method will be logged and ignored, and will not result in task failure.
*
* @param latestCommittedOffsets Latest committed source-offsets produced by this task. For each key ({SourceRecord.sourcePartition}),
* the associated value ({SourceRecord.sourceOffset}) represents the most recent committed source-offset.
* This ensures that all preceding sourceOffsets, as delivered by task.poll() for the same sourcePartition,
* have also been committed.
* @throws InterruptedException
*/
public void commit(Map<Map<String, Object>, Map<String, Object>> latestCommittedOffsets) throws InterruptedException {
this.commit();
} |
...