DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
This page is meant as a template for writing a KIP. To create a KIP choose Tools->Copy on this page and modify with your content and replace the heading with the next KIP number and a description of your issue. Replace anything in italics with your own description.
Status
Current state: "DraftUnder discussion"
Discussion thread: here
JIRA: here
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
...
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 that were committed. These offsets are in the order they were 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 (ordered wrt {SourceRecord.sourcePartition}).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();
} |
...