Page History

...

This change in behavior has implications for the consumer's offset reset policy, which defines what the consumer should do if its fetch offset becomes out of range. With this KIP, the only case in which this is possible is if the consumer fetches from an offset earlier than the log start offset. By opting into an offset reset policy, the user allows for automatic adjustments to the fetch position, so we take advantage of this to to reset the offset as precisely as possible when log truncation is detected. In some pathological cases (e.g. multiple consecutive unclean leader elections), we may not be able to find the exact offset, but we should be able to get close by finding the starting offset of the next largest epoch that the leader is aware of. We propose in this KIP to change the behavior for both the "earliest" and "latest" reset modes to do this automatically as long as the message format supports lookup by leader epoch.   The consumer will log a message to indicate that the truncation was detected, but will reset the position automatically. 

If a user is not using an auto reset option, we will raise a LogTruncationException from poll() when log truncation is detected. This gives users the ability to reset state if needed or revert changes to downstream systems. The exception will include the partitions that were truncated and the offset of divergence as found above. This gives applications the ability to execute any logic to revert changes if needed and rewind the fetch offset. Users must handle this exception and reset the position of the consumer if no auto reset policy is enabled.

...

/**
 * In the even of unclean leader election, the log will be truncated,
 * previously committed data will be lost, and new data will be written
 * over these offsets. When this happens, the consumer will detect the 
 * truncation and raise this exception (if no automatic reset policy 
 * has been defined) with the first offset to diverge from what the 
 * consumer read.
 */ 
class LogTruncationException extends OffsetOutOfRangeException {
  /**
   * Get the truncation offsets offsetfor ifthe onepartitions couldwhich bewere foundtruncated. 
   * This is the first
   * offset which is known to diverge from what the consumer read.
   */	
  Map<TopicPartition, Optional<OffsetAndMetadata>OffsetAndMetadata> truncationOffsettruncationOffsets();
}

We will also add a new retriable error code exceptions for the UNKNOWN_LEADER_EPOCH and FENCED_LEADER_EPOCH error codecodes:

class UnknownLeaderEpochException extends RetriableException {}


class FencedLeaderEpochException extends RetriableException {}

...