[ashwinhs]

Description

Kafka source records ahead of the consumer clock now use the current time for origination time instead of an old cached timestamp. kafka_timestamp keeps the record's value.

Tested with Kafka 3.9.1 and OpenSearch 3.9.0: one normal record, then five records 5s ahead after a 20s wait.

	Before	After
kafka_timestamp error	25-26s	0s
Avg EndToEndLatency	17.7s	0.49s
Max EndToEndLatency	22.2s	0.99s

numberOfInvalidTimeStamps still counts these records.

Issues Resolved

Resolves #7158

Check List

• New functionality includes testing.
• New functionality has a documentation issue. Please link to it in this PR.
  • New functionality has javadoc added
• Commits are signed with a real name per the DCO

By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.

[github-actions]

PR Reviewer Guide 🔍

(Review updated until commit 73691cd)

Here are some key observations to aid the review process:

🧪 PR contains tests

🔒 No security concerns identified

✅ No TODO sections

🔀 No multiple PR themes

⚡ No major issues detected

[github-actions]

PR Code Suggestions ✨

Explore these optional code suggestions:

Category	Suggestion	Impact
General	Keep metadata timestamp consistent with clamping Setting kafka_timestamp directly from consumerRecord.timestamp() bypasses the invalid-timestamp metric increment and lets future-dated timestamps leak into event metadata, which contradicts the clamping applied to externalOriginationTime. Keep the attribute consistent with the clamped value, or at least call getRecordTimeStamp so the metric still fires. data-prepper-plugins/kafka-plugins/src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java [503] final long receivedTimeStamp = getRecordTimeStamp(consumerRecord, now.toEpochMilli()); -eventMetadata.setAttribute("kafka_timestamp", consumerRecord.timestamp()); +eventMetadata.setAttribute("kafka_timestamp", receivedTimeStamp); Suggestion importance[1-10]: 2 __ Why: The suggestion contradicts the explicit intent of the PR, which is to preserve the original kafka_timestamp from the record while clamping only the externalOriginationTime. The test testFutureRecordKeepsKafkaTimestampAndClampsOriginationTime confirms this is the desired behavior.	Low

[srikanthpadakanti]

Thanks for the fix and the clear numbers, @ashwinhs! One small thing: Data Prepper tests use Hamcrest instead of JUnit assertTrue. Could you change these two lines to assertThat(..., lessThan(futureTimestamp))?

[ashwinhs]

Done @srikanthpadakanti thanks. The Kafka CI integration failures also happen on main. I'll open a separate PR to fix them.

[srikanthpadakanti]

Thanks @ashwinhs, looks good.

[github-actions]

Persistent review updated to latest commit 73691cd

[srikanthpadakanti]

LGTM