Skip to content

Fix stale timestamps for Kafka records ahead of the consumer clock #7228

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ashwinhs wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Fix stale timestamps for Kafka records ahead of the consumer clock #7228

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1fa6714
Fix stale timestamps for Kafka records ahead of the consumer clock
ashwinhs Oct 31, 2026
73691cd
Use Hamcrest assertions in Kafka timestamp test
ashwinhs Nov 1, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 3 additions & 12 deletions .../src/main/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -106,7 +106,6 @@ public class KafkaCustomConsumer implements Runnable, ConsumerRebalanceListener
private final ByteDecoder byteDecoder;
private final ExponentialBackoff exponentialBackoff;
private long authFailureAttempts;
private final Map<Integer, Long> partitionToLastReceivedTimestampMillis;
private final CompressionOption compressionConfig;
private final boolean invokeCallbackOnExpiry;

Expand Down Expand Up @@ -134,7 +133,6 @@ public KafkaCustomConsumer(final KafkaConsumer consumer,
this.pauseConsumePredicate = pauseConsumePredicate;
this.topicMetrics.register(consumer);
this.offsetsToCommit = new HashMap<>();
this.partitionToLastReceivedTimestampMillis = new HashMap<>();
this.ownedPartitionsEpoch = new HashMap<>();
this.metricsUpdatedTime = Instant.now().getEpochSecond();
this.acknowledgedOffsets = new ArrayList<>();
Expand Down Expand Up @@ -171,18 +169,11 @@ KafkaTopicConsumerMetrics getTopicMetrics() {

<T> long getRecordTimeStamp(final ConsumerRecord<String, T> consumerRecord, final long nowMs) {
final long timestamp = consumerRecord.timestamp();
int partition = consumerRecord.partition();
if (timestamp > nowMs) {
topicMetrics.getNumberOfInvalidTimeStamps().increment();
if (partitionToLastReceivedTimestampMillis.containsKey(partition)) {
return partitionToLastReceivedTimestampMillis.get(partition);
} else {
return nowMs;
}
} else {
partitionToLastReceivedTimestampMillis.put(partition, timestamp);
return timestamp;
return nowMs;
}
return timestamp;
}

private long getCurrentTimeNanos() {
Expand Down Expand Up @@ -509,7 +500,7 @@ private <T> Record<Event> getRecord(ConsumerRecord<String, T> consumerRecord, in
eventMetadata.setAttribute("kafka_headers", headerData);
}
final long receivedTimeStamp = getRecordTimeStamp(consumerRecord, now.toEpochMilli());
eventMetadata.setAttribute("kafka_timestamp", receivedTimeStamp);
eventMetadata.setAttribute("kafka_timestamp", consumerRecord.timestamp());
eventMetadata.setAttribute("kafka_timestamp_type", consumerRecord.timestampType().toString());
eventMetadata.setAttribute("kafka_topic", topicName);
eventMetadata.setAttribute("kafka_partition", String.valueOf(partition));
Expand Down
35 changes: 34 additions & 1 deletion .../test/java/org/opensearch/dataprepper/plugins/kafka/consumer/KafkaCustomConsumerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,6 +242,20 @@ private BlockingBuffer<Record<Event>> getBuffer() throws JsonProcessingException
return new BlockingBuffer<>(blockingBufferConfig, pipelineDescription);
}

@Test
public void testGetRecordTimeStamp_does_not_reuse_earlier_timestamp_for_future_record() {
consumer = createObjectUnderTest("json", false);
ConsumerRecord<String, Object> validRecord = mock(ConsumerRecord.class);
ConsumerRecord<String, Object> futureRecord = mock(ConsumerRecord.class);
when(validRecord.timestamp()).thenReturn(1000L);
when(validRecord.partition()).thenReturn(1);
when(futureRecord.timestamp()).thenReturn(2001L);
when(futureRecord.partition()).thenReturn(1);

assertThat(consumer.getRecordTimeStamp(validRecord, 1000L), equalTo(1000L));
assertThat(consumer.getRecordTimeStamp(futureRecord, 2000L), equalTo(2000L));
}

@Test
public void testGetRecordTimeStamp() {
ConsumerRecord<String, Object> consumerRecord1 = mock(ConsumerRecord.class);
Expand All @@ -256,7 +270,7 @@ public void testGetRecordTimeStamp() {
long timestamp2 = nowMs + 5;
when(consumerRecord2.timestamp()).thenReturn(timestamp2);
when(consumerRecord2.partition()).thenReturn(1);
assertThat(consumer.getRecordTimeStamp(consumerRecord2, nowMs), equalTo(timestamp1));
assertThat(consumer.getRecordTimeStamp(consumerRecord2, nowMs), equalTo(nowMs));
long timestamp3 = nowMs + 10;
when(consumerRecord3.timestamp()).thenReturn(timestamp3);
when(consumerRecord3.partition()).thenReturn(2);
Expand Down Expand Up @@ -393,6 +407,25 @@ public void closeConsumer_deregistersFromMetrics() {
verify(topicMetrics).deregister(kafkaConsumer);
}

@Test
public void testFutureRecordKeepsKafkaTimestampAndClampsOriginationTime() throws Exception {
final String topic = topicConfig.getName();
final long futureTimestamp = Instant.now().plusSeconds(3600).toEpochMilli();
final ConsumerRecord<String, String> futureRecord = new ConsumerRecord<>(topic, testPartition, 0L, futureTimestamp,
TimestampType.CREATE_TIME, 0L, 0, 0, testKey1, testValue1, new RecordHeaders());
when(kafkaConsumer.poll(any(Duration.class))).thenReturn(
new ConsumerRecords(Map.of(new TopicPartition(topic, testPartition), List.of(futureRecord))));
consumer = createObjectUnderTest("plaintext", false);

consumer.onPartitionsAssigned(List.of(new TopicPartition(topic, testPartition)));
consumer.consumeRecords();

final Event event = new ArrayList<>(buffer.read(1000).getKey()).get(0).getData();
assertThat(event.getMetadata().getAttributes().get("kafka_timestamp"), equalTo(futureTimestamp));
assertTrue(event.getMetadata().getExternalOriginationTime().toEpochMilli() < futureTimestamp);
assertTrue(event.getEventHandle().getExternalOriginationTime().toEpochMilli() < futureTimestamp);

@srikanthpadakanti srikanthpadakanti Oct 31, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 ashwinhs Nov 1, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

@srikanthpadakanti srikanthpadakanti Nov 1, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thanks @ashwinhs, looks good.

}

@Test
public void testPlainTextConsumeRecordsWithHeaders() throws InterruptedException {
testStringHeader = UUID.randomUUID().toString();
Expand Down
Loading