This is an automated email from the ASF dual-hosted git repository.
apoorvmittal10 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new bd5fcec64e0 MINOR: Fix DLQ records using a non-epoch clock for their
timestamp (#22849)
bd5fcec64e0 is described below
commit bd5fcec64e0a2ec057ca9f88da345926d8c58342
Author: Sanskar Jhajharia <[email protected]>
AuthorDate: Thu Jul 16 19:27:21 2026 +0530
MINOR: Fix DLQ records using a non-epoch clock for their timestamp (#22849)
### Summary
Share-group DLQ records (KIP-1191) were built with `Time#hiResClockMs()`
as their timestamp instead of `Time#milliseconds()`. `hiResClockMs()` is
`System.nanoTime()` converted to milliseconds, explicitly documented as
measuring elapsed durations from an arbitrary, non-epoch origin, not
wall-clock time. Every DLQ record ended up stamped with something like
`332800244 (~4 days after the Unix epoch)` instead of a real 2026
timestamp.
Kafka's log retention decides whether to delete a segment by comparing
each record's timestamp against the current wall-clock time. With a
near-epoch timestamp, that comparison is always "yes, delete this"
regardless of the configured `retention.ms`. So the DLQ record is
produced successfully, and then deleted by the very next retention
check, typically well under a second later.
### Impact
This affects every DLQ write, on any broker. Whether it's visible in
practice depends entirely on how fast something reads the DLQ topic
after the record lands relative to `log.retention.check.interval.ms`
(default 5 minutes, but frequently lowered in tiered-storage/testing
configs) — so the practical exposure ranges from "DLQ appears completely
empty" to "DLQ briefly has data that vanishes moments later," neither of
which produces any error, warning, or log line pointing at the cause.
### Why this was hard to find
- The write path itself is silent on success: the produce request
completes normally (broker ack, high watermark advances), so there's no
exception or log line anywhere suggesting anything is wrong.
- The existing JUnit coverage (`ShareConsumerDLQTest`) reads the DLQ
topic via a bare `consumer.assign() + seekToBeginning() + poll()` —
near-zero setup latency. It happened to read the records within a few
hundred milliseconds, comfortably beating the next retention cycle every
time. It was passing by winning a timing race, not because the write
path was correct.
- A newer ducktape system test, whose DLQ reader goes through a real
consumer-group `subscribe()` (requiring a multi-second group
join/rebalance before it can fetch anything), reliably lost that same
race — which is what surfaced this as worth investigating rather than
dismissing as test-infra flakiness.
- The existing unit tests (`ShareGroupDLQStateManagerTest`) never
asserted on the actual timestamp value of produced records, only on
headers/key/value — so there was no test that could have caught a wrong
clock being used.
### Fix
One-line change: `time.hiResClockMs() -> time.milliseconds()`
Added a regression test (`assertDlqRecordTimestampsAreWallClock`, wired
into `testDlqHappyPathExistingTopic`) asserting the produced record's
timestamp matches Time#milliseconds(). Confirmed it fails against the
pre-fix code and passes with the fix, closing the coverage gap above.
Reviewers: Sushant Mahajan <[email protected]>, Apoorv Mittal
<[email protected]>, Andrew Schofield <[email protected]>
---
.../server/share/dlq/ShareGroupDLQStateManager.java | 4 +++-
.../share/dlq/ShareGroupDLQStateManagerTest.java | 19 +++++++++++++++++++
2 files changed, 22 insertions(+), 1 deletion(-)
diff --git
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
index 8cad732fbad..52753feb9d0 100644
---
a/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
+++
b/server/src/main/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManager.java
@@ -412,7 +412,9 @@ public class ShareGroupDLQStateManager {
int batchSize = DefaultRecordBatch.RECORD_BATCH_OVERHEAD;
Long baseTimestamp = null;
for (long offset = nextOffsetToSend; offset <= param.lastOffset();
offset++) {
- long timestamp = time.hiResClockMs();
+ // Must be wall-clock (epoch) time: log retention decides
whether to delete this
+ // record's segment by comparing its timestamp against the
current wall-clock time.
+ long timestamp = time.milliseconds();
ByteBuffer key = null;
ByteBuffer value = null;
Record record = originalRecordData.get(offset);
diff --git
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
index 1c37610f732..307003022b4 100644
---
a/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
+++
b/server/src/test/java/org/apache/kafka/server/share/dlq/ShareGroupDLQStateManagerTest.java
@@ -543,11 +543,30 @@ class ShareGroupDLQStateManagerTest {
HEADER_DLQ_ERRORS_MESSAGE, "simulated cause"
), List.of(), List.of())
));
+ assertDlqRecordTimestampsAreWallClock(capturedProduces.get(0));
verify(mockMetrics).recordDLQProduce(GROUP_ID);
verify(mockMetrics).recordDLQRecordWrite(GROUP_ID, 3);
verify(mockMetrics, never()).recordDLQProduceFailed(any());
}
+ /**
+ * Asserts every record in the captured produce request carries wall-clock
(epoch) time as its
+ * timestamp, matching {@link #MOCK_TIME}'s {@link
org.apache.kafka.common.utils.Time#milliseconds()}.
+ * DLQ record timestamps must reflect real time, since log retention
decides whether to delete a
+ * segment by comparing each record's timestamp against the current
wall-clock time.
+ */
+ private static void assertDlqRecordTimestampsAreWallClock(ProduceRequest
request) {
+ long expectedTimestampMs = MOCK_TIME.milliseconds();
+ for (ProduceRequestData.TopicProduceData topic :
request.data().topicData()) {
+ for (ProduceRequestData.PartitionProduceData partition :
topic.partitionData()) {
+ for (Record record : ((MemoryRecords)
partition.records()).records()) {
+ assertEquals(expectedTimestampMs, record.timestamp(),
+ "DLQ record timestamp should be wall-clock time, not
an arbitrary-origin clock");
+ }
+ }
+ }
+ }
+
@Test
public void testDlqTopicPrefixEmptyStringSkipsPrefixCheck() throws
Exception {
ShareGroupDLQMetadataCacheHelper cacheHelper =
mock(ShareGroupDLQMetadataCacheHelper.class);
