...

Authors Mehari Beyene, Divij Vaidya

Status

Current state: Under Accepted

Discussion thread: here

Discussion Vote thread: here

JIRA: KAFKA-14991

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

...

There will be no change to existing public interfaces. However, when a producer's record is rejected due to this new the timestamp validation, we will return error code 32 (INVALID_TIMESTAMP) with the error message "Timestamp [record timestamp] of the message with offset [record offset] is ahead of the broker's current time."out of range. The timestamp should be within [ [brokerTime - log.message.timestamp.before.max.ms], [brokerTime + log.message.timestamp.after.max.ms:] ]."

Note that the error message will remain the same as the current timestamp validation, except that we will be introducing two new configurations: `log.message.timestamp.before.max.ms` and `log.message.timestamp.after.max.ms`. Additionally, We are also introducing two new configurations and deprecating the configuration log.message.timestamp.difference.max.ms as will be deprecated. Details of these configuration changes are discussed in the Proposed Changes section.

Proposed Changes

We will introduce two new configurations: log.message.timestamp.before.max.ms and log.message.timestamp.after.max.ms. Additionally, we will deprecate the existing configuration log.message.timestamp.difference.max.ms.

...

log.message.timestamp.after.max.ms: This configuration defines that the message timestamp must always can not be later than or equal to the broker's timestamp, with a maximum allowable difference determined by the value of log.message.timestamp.after.max.ms.

  - Type: long
  - Default: 36000009223372036854775807
  - Valid Values: [0,...]
  - Update Mode: cluster-wide

Note that the default value for of log.message.timestamp.after.max.ms is set to 3600000 milliseconds (one hour). Considering potential clock drift issues that the broker may encounter with the time synchronization service of the underlying operating system, we propose the default value of this configuration to be one hour. Time synchronization services like NTP and PTP are capable of fixing drift in the order of milliseconds. Therefore, assuming a one-hour threshold should be sufficient to accommodate all clock drift caseskept the same as log.message.timestamp.difference.max.ms to maintain backward compatibility and not introduce a breaking change for clients that are sending messages with future timestamps. However, in a future major version bump, we will change this default value to 3600000 milliseconds (one hour). To prepare users for this change in the future, we will add a WARN-level log when we encounter messages with future timestamps that are ahead of the proposed one-hour threshold.

Timestamp Validation Logic

...

if message.timestamp <= broker.timestamp:
    time_difference = broker.timestamp - message.timestamp
    if time_difference <= log.message.timestamp.before.max.ms:
        # Validation passed for timestamp before
        //# Proceed with further logic
    else:
        # Validation failed for timestamp before
        return Error code 32 (INVALID_TIMESTAMP)
else if message.timestamp > broker.timestamp:
    time_difference = message.timestamp - broker.timestamp
    if time_difference <= log.message.timestamp.after.max.ms:
        # Validation passed for timestamp after
        //# Proceed with further logic
    else:
        # Validation failed for timestamp after
        return Error code 32 (INVALID_TIMESTAMP)

...

Compatibility, Deprecation, and Migration Plan

• The configuration log.message.timestamp.difference.max.ms will be deprecated.
• The semantics of the configuration regarding time differences in the past and future were not explicitly defined with the configuration log.message.timestamp.difference.max.ms. With this KIP, message timestamps with past and future values are now validated using different default values, defined by the configurations log.message.timestamp.before.max.ms and log.message.timestamp.after.max.ms respectively.
• After this KIP messages with future timestamps are validated using the new configuration log.message.timestamp.after.max.ms. This will result in for some messages to be rejected that were previously accepted. We consider this an acceptable breaking change.

There are no other changes to public interfaces that will impact clients.

...

• Lack of Flexibility: Users may have use cases that require extending the future timestamp validation beyond the default one hour, and the hardcoded constant value restricts this flexibility.
• Compatibility Issues: The hardcoded constant value of one hour can create compatibility issues when upgrading existing clusters or integrating with existing producers.

...