You are viewing an old version of this page. View the current version.

Compare with Current View Page History

Version 1 Next »

Status

Current state: Under Discussion

Discussion thread: here [Change the link from the KIP proposal email archive to your own email thread]

JIRA: here [Change the link from KAFKA-1 to your own ticket]

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

MirrorMaker 2.0 (KIP-382) needs to know the downstream offsets of replicated records in order to provide cross-cluster offset translation. Currently, WorkerSourceTask receives this information from KafkaProducer but throws it away. It's possible that other Connectors may benefit from this change, e.g. see KIP-381, which also proposes to notify SourceTasks of ACK'd records. In particular, this proposal makes it possible to distinguish when records have been durably stored vs when they have been skipped altogether by a SourceConnector.

Public Interfaces

A new callback method recordLogged() will be added to SourceTask:


public abstract class SourceTask implements Task {
---%<---
public void recordLogged(SourceRecord sourceRecord, RecordMetadata recordMetadata) {
  // nop
}
---%<---


Proposed Changes

Currently, SourceTask includes a commitRecord() callback, which is invoked under these conditions:

  • record is ACK'd from producer
  • SourceRecord is filtered out from transformation chain, and thus is never sent to Kafka
  • SourceTask gives up after retries and skips the SourceRecord.

The new method recordLogged() will be invoked only when a record is ACK'd, which implies the record was not filtered and was not skipped.

After an ACK, WorkerSourceTask currently logs MetadataRecord.offset() and partition() before invoking the commitRecord callback. I propose to also invoke the new recordLogged() callback as well:


task.recordLogged(preTransformRecord, recordMetadata);
task.commitRecord(preTransformRecord);


Compatibility, Deprecation, and Migration Plan

This is a new callback and won't affect existing code beyond an additional no-op function call. 


Rejected Alternatives

  • We could call it recordAcked(), but I think recordLogged() better captures what is being ACK'd.
  • SinkRecord has offset and partition fields, so we could potentially use recordLogged(SinkRecord), but this would be a confusing abuse of SinkRecord's semantics. In particular, it would be weird to have SinkRecord in the SourceTask interface.
  • We could add kafkaOffset and kafkaPartition fields to SourceRecord, which would be null prior to ACK and then filled in by WorkerSourceTask after ACK. However, this breaks the "value class" semantics of SourceRecord.
  • We could extend SourceRecord with AckedSourceRecord or LoggedSourceRecord, but this seems overkill.
  • We could avoid using RecordMetadata in the API, and instead include kafkaOffset, kafkaPartition, kafkaTopic, and kafkaTimestamp as parameters to recordLogged(). But this is a lot of parameters, and RecordMatadata is already in the clients API.
  • We could pass just the kafkaOffset to recordLogged(), but an offset is mostly meaningless without an associated partition. We could include both kafkaOffset and kafkaPartition, but this doesn't account for transformations that may change the downstream topic name. We could include kafkaTopic as well, but then we might as well include the entire RecordMetadata.
  • No labels