Status

Current state: Under discussion. Accepted

Discussion thread: TBDhttp://search-hadoop.com/m/Kafka/uyzND1DqvOK1jiI281?subj=+DISCUSS+KIP+192+Provide+cleaner+semantics+when+idempotence+is+enabled

JIRA:  https://issues.apache.org/jira/browse/KAFKA-5793 and https://issues.apache.org/jira/browse/KAFKA-5794

Release Version : Update to the RecordMetadata class and ProduceResponse are in 1.0.0. The new values for enable.idempotence in the ProducerConfig and the updates to the TopicMetadata in MetadataResponse has been postponed to a future release.

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

...

1. Kafka Exactly Once - Solving the problem of spurious OutOfOrderSequence errors
2. Kafka Exactly Once - Dealing with older message formats when idempotence is enabled

Public Interfaces

RecordMetadata (in 1.0.0)

With the changes in Kafka Exactly Once - Solving the problem of spurious OutOfOrderSequence errors, the broker may return a new DUPLICATE_SEQUENCE error code in some cases where a duplicate is detected but the metadata for the existing batch isn't cached in memory. When the producer receives this error, it is considered a success, but will not have the offset and timestamp information for the appended records. If the user then calls RecordMetadata.offset or RecordMetadata.timestamp they will receive a new RecordMetadataNotAvailableException.To help identify this state, we add  hasOffset and add hasTimestamp methods to the RecordMetadata.

package org.apache.clients.producer;
 
public final class RecordMetadata {
 
  /** 
   * The RecordMetadataNotAvailableException may be thrown if the broker was unable  Indicates whether the record metadata includes the offset.
   * @return true if the offset is available, false otherwise.
   */
  public boolean hasOffset();
  
  /**
   * toIndicates returnwhether the necessaryrecord metadata toincludes the clienttimestamp.
   */
  public long offset() throws RecordMetadataNotAvailableException;
   @return true if the timestamp is available, false otherwise.
   */ 
  public longboolean timestamphasTimestamp() throws RecordMetadataNotAvailableException;
 
}

TopicMetadataResponse

...

// TopicMetadataV3
 
TopicMetadata => TopicErrorCode
                 Topic
                 IsInternal
				 MessageFormatVersion
				 MaxMessageBytes
				 [PartitionMetadata]
 TopicErrorCode => int16
 Topic => String
 IsInternal => Boolean
 MessageFormatVersion => int8 (NEW)
 MaxMessageBytes => int32 (NEW)
 PartitionMetadata => PartitionMetadataV2

ProduceResponse (in 1.0.0)

We add a logStartOffset field to the produce response to help the producer identify when producer state has been lost due to retention time elapsing. See Kafka Exactly Once - Solving the problem of spurious OutOfOrderSequence errors for a precise description of how this will be used.

...

As part of these changes, we will deprecate the true and false options for enable.idempotence by logging a warning if these are used.

Finally, the For applications which care about always receiving the offset and timestamp of produced records, there is a greater chance that these will not be available when idempotence is enabled (for instance a broker bounce would lose some of the cached record metadata, and if an internal producer retry resulted in a duplicate of a record which is dropped from the cache, the record metadata would not be returned for this record, even though the append is successful). Such applications should now check the new RecordMetadata.hasOffset and RecordMetadtata.hasTimestamp methods before using the values returned from the RecordMetadata.offset and RecordMetadata.timestamp methods may throw the new RecordMetadataNotAvailableException if the broker was unable to return the metadata for the record in question. 

Rejected Alternatives

This KIP contains changes to fix existing problems or clarify existing behavior. As such, there are not too many options for making these improvements within the existing solutions. 

...