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 all commits
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
36 changes: 35 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 @@ -75,6 +75,7 @@
import static org.hamcrest.CoreMatchers.notNullValue;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.hasEntry;
import static org.hamcrest.Matchers.lessThan;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertEquals;
Expand Down Expand Up @@ -242,6 +243,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 +271,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 +408,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));
assertThat(event.getMetadata().getExternalOriginationTime().toEpochMilli(), lessThan(futureTimestamp));
assertThat(event.getEventHandle().getExternalOriginationTime().toEpochMilli(), lessThan(futureTimestamp));
}

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