This is an automated email from the ASF dual-hosted git repository.
xiangfu0 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git
The following commit(s) were added to refs/heads/master by this push:
new e0210aaf253 Report NOT_CALCULATED for uncomputable consuming-segment
lag (#18836) (#18971)
e0210aaf253 is described below
commit e0210aaf2530fc123aa48f7560174c59d10aaec5
Author: Xiang Fu <[email protected]>
AuthorDate: Sat Jul 11 07:34:37 2026 +0200
Report NOT_CALCULATED for uncomputable consuming-segment lag (#18836)
(#18971)
The consuming-segments panel's "Max Partition Availability Lag (ms)" showed
an epoch-sized value when a record's upstream ingestion timestamp was
missing
or invalid. The availability lag was computed as
`lastProcessedTimeMs - recordIngestionTimeMs` without validating the
ingestion
time, so a record with no timestamp (Kafka NO_TIMESTAMP = -1, an unset
Long.MIN_VALUE, or 0) turned the subtraction into a nonsensical ~now-sized
lag
that leaked into the MAX_RECORD_AVAILABILITY_LAG_MS metric and the UI.
Guard the availability-lag computation on `getRecordIngestionTimeMs() > 0`
in
all three StreamMetadataProvider implementations (kafka-3.0, kafka-4.0,
kinesis) so an invalid ingestion time reports the SPI's
PartitionLagState.NOT_CALCULATED sentinel instead. Also replace the
hard-coded
"UNKNOWN" fallback (offset and availability lag) with NOT_CALCULATED to
align
with the SPI-defined sentinel; downstream consumers
(RealtimeConsumerMonitor,
UI) already treat it as "no value".
Add per-module regression tests covering valid, unset, NO_TIMESTAMP (-1),
and
0 ingestion times, plus the offset-lag fallback.
---
.../kafka30/KafkaStreamMetadataProvider.java | 12 ++--
.../kafka30/KafkaStreamMetadataProviderTest.java | 65 ++++++++++++++++++++++
.../kafka40/KafkaStreamMetadataProvider.java | 12 ++--
.../kafka40/KafkaStreamMetadataProviderTest.java | 65 ++++++++++++++++++++++
.../kinesis/KinesisStreamMetadataProvider.java | 10 +++-
.../kinesis/KinesisStreamMetadataProviderTest.java | 48 ++++++++++++++++
6 files changed, 201 insertions(+), 11 deletions(-)
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java
index 130a34d3743..84d4a7ac050 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/main/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProvider.java
@@ -235,17 +235,21 @@ public class KafkaStreamMetadataProvider extends
KafkaPartitionLevelConnectionHa
// Compute records-lag
StreamPartitionMsgOffset currentOffset =
partitionState.getCurrentOffset();
StreamPartitionMsgOffset upstreamLatest =
partitionState.getUpstreamLatestOffset();
- String offsetLagString = "UNKNOWN";
+ String offsetLagString = PartitionLagState.NOT_CALCULATED;
if (currentOffset instanceof LongMsgOffset && upstreamLatest instanceof
LongMsgOffset) {
long offsetLag = ((LongMsgOffset) upstreamLatest).getOffset() -
((LongMsgOffset) currentOffset).getOffset();
offsetLagString = String.valueOf(offsetLag);
}
- // Compute record-availability
- String availabilityLagMs = "UNKNOWN";
+ // Compute record-availability. Only when both the last-processed
wall-clock time and the record's upstream
+ // ingestion time are valid; otherwise a missing/invalid ingestion time
(e.g. records without a Kafka
+ // timestamp) would turn the subtraction into an epoch-sized value that
leaks to the metric and UI. Keep the
+ // NOT_CALCULATED sentinel in that case so the lag is reported as
not-calculated rather than a bogus number.
+ String availabilityLagMs = PartitionLagState.NOT_CALCULATED;
StreamMessageMetadata lastProcessedMessageMetadata =
partitionState.getLastProcessedRowMetadata();
- if (lastProcessedMessageMetadata != null &&
partitionState.getLastProcessedTimeMs() > 0) {
+ if (lastProcessedMessageMetadata != null &&
partitionState.getLastProcessedTimeMs() > 0
+ && lastProcessedMessageMetadata.getRecordIngestionTimeMs() > 0) {
long availabilityLag =
partitionState.getLastProcessedTimeMs() -
lastProcessedMessageMetadata.getRecordIngestionTimeMs();
availabilityLagMs = String.valueOf(availabilityLag);
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProviderTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProviderTest.java
index 973f338bbb7..5f3d0506b88 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProviderTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-3.0/src/test/java/org/apache/pinot/plugin/stream/kafka30/KafkaStreamMetadataProviderTest.java
@@ -30,11 +30,14 @@ import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.pinot.spi.stream.ConsumerPartitionState;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+import org.apache.pinot.spi.stream.PartitionLagState;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
+import org.apache.pinot.spi.stream.StreamMessageMetadata;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.testng.annotations.Test;
@@ -111,6 +114,68 @@ public class KafkaStreamMetadataProviderTest {
}
}
+ @Test
+ public void testGetCurrentPartitionLagStateHandlesInvalidIngestionTime()
+ throws Exception {
+ String topicName = "asset";
+ Consumer<Bytes, Bytes> consumer = mockConsumer(topicName, 1);
+ MOCK_CONSUMER.set(consumer);
+ try {
+ StreamConfig streamConfig = getStreamConfig(topicName);
+ try (KafkaStreamMetadataProvider provider = new
MockKafkaStreamMetadataProvider("client", streamConfig)) {
+ long lastProcessedTimeMs = 1_700_000_100_000L;
+
+ // Record with a valid upstream ingestion time yields a numeric
availability lag.
+ StreamMessageMetadata validMetadata = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .setRecordIngestionTimeMs(lastProcessedTimeMs - 1000L)
+ .build();
+ // Records whose ingestion time is missing/invalid: unset (Builder
default Long.MIN_VALUE), Kafka's
+ // NO_TIMESTAMP (-1), and epoch 0 (the exact boundary of the > 0
guard). These stand in for a topic that is
+ // unreachable/timing out or produces records without a timestamp.
+ StreamMessageMetadata unsetIngestionTime = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .build();
+ StreamMessageMetadata noTimestampIngestionTime = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .setRecordIngestionTimeMs(-1L)
+ .build();
+ StreamMessageMetadata zeroIngestionTime = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .setRecordIngestionTimeMs(0L)
+ .build();
+
+ Map<String, ConsumerPartitionState> stateMap = new HashMap<>();
+ stateMap.put("0", new ConsumerPartitionState("0", new
LongMsgOffset(5), lastProcessedTimeMs,
+ new LongMsgOffset(10), validMetadata));
+ stateMap.put("1", new ConsumerPartitionState("1", new
LongMsgOffset(5), lastProcessedTimeMs,
+ new LongMsgOffset(10), unsetIngestionTime));
+ stateMap.put("2", new ConsumerPartitionState("2", new
LongMsgOffset(5), lastProcessedTimeMs,
+ new LongMsgOffset(10), noTimestampIngestionTime));
+ stateMap.put("3", new ConsumerPartitionState("3", new
LongMsgOffset(5), lastProcessedTimeMs,
+ new LongMsgOffset(10), zeroIngestionTime));
+ // Partition with an unknown upstream offset exercises the offset-lag
fallback.
+ stateMap.put("4", new ConsumerPartitionState("4", new
LongMsgOffset(5), lastProcessedTimeMs,
+ null, validMetadata));
+
+ Map<String, PartitionLagState> lagState =
provider.getCurrentPartitionLagState(stateMap);
+
+ // Offset lag: numeric when both offsets are known, NOT_CALCULATED
when the upstream offset is unavailable.
+ assertEquals(lagState.get("0").getRecordsLag(), "5");
+ assertEquals(lagState.get("4").getRecordsLag(),
PartitionLagState.NOT_CALCULATED);
+ // Availability lag: numeric for a valid ingestion time,
NOT_CALCULATED for every invalid one.
+ // Regression for issue #18836: an invalid ingestion time must not
leak an epoch-sized value
+ // (lastProcessedTimeMs - Long.MIN_VALUE, or lastProcessedTimeMs -
(-1) ~= now).
+ assertEquals(lagState.get("0").getAvailabilityLagMs(), "1000");
+ assertEquals(lagState.get("1").getAvailabilityLagMs(),
PartitionLagState.NOT_CALCULATED);
+ assertEquals(lagState.get("2").getAvailabilityLagMs(),
PartitionLagState.NOT_CALCULATED);
+ assertEquals(lagState.get("3").getAvailabilityLagMs(),
PartitionLagState.NOT_CALCULATED);
+ }
+ } finally {
+ MOCK_CONSUMER.remove();
+ }
+ }
+
private static StreamConfig getStreamConfig(String topicName) {
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put("streamType", "kafka");
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java
index 8db789a79cb..63305beb463 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/main/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProvider.java
@@ -235,17 +235,21 @@ public class KafkaStreamMetadataProvider extends
KafkaPartitionLevelConnectionHa
// Compute records-lag
StreamPartitionMsgOffset currentOffset =
partitionState.getCurrentOffset();
StreamPartitionMsgOffset upstreamLatest =
partitionState.getUpstreamLatestOffset();
- String offsetLagString = "UNKNOWN";
+ String offsetLagString = PartitionLagState.NOT_CALCULATED;
if (currentOffset instanceof LongMsgOffset && upstreamLatest instanceof
LongMsgOffset) {
long offsetLag = ((LongMsgOffset) upstreamLatest).getOffset() -
((LongMsgOffset) currentOffset).getOffset();
offsetLagString = String.valueOf(offsetLag);
}
- // Compute record-availability
- String availabilityLagMs = "UNKNOWN";
+ // Compute record-availability. Only when both the last-processed
wall-clock time and the record's upstream
+ // ingestion time are valid; otherwise a missing/invalid ingestion time
(e.g. records without a Kafka
+ // timestamp) would turn the subtraction into an epoch-sized value that
leaks to the metric and UI. Keep the
+ // NOT_CALCULATED sentinel in that case so the lag is reported as
not-calculated rather than a bogus number.
+ String availabilityLagMs = PartitionLagState.NOT_CALCULATED;
StreamMessageMetadata lastProcessedMessageMetadata =
partitionState.getLastProcessedRowMetadata();
- if (lastProcessedMessageMetadata != null &&
partitionState.getLastProcessedTimeMs() > 0) {
+ if (lastProcessedMessageMetadata != null &&
partitionState.getLastProcessedTimeMs() > 0
+ && lastProcessedMessageMetadata.getRecordIngestionTimeMs() > 0) {
long availabilityLag =
partitionState.getLastProcessedTimeMs() -
lastProcessedMessageMetadata.getRecordIngestionTimeMs();
availabilityLagMs = String.valueOf(availabilityLag);
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProviderTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProviderTest.java
index f3a34b57fe7..267e288ecc8 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProviderTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kafka-4.0/src/test/java/org/apache/pinot/plugin/stream/kafka40/KafkaStreamMetadataProviderTest.java
@@ -30,11 +30,14 @@ import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.common.PartitionInfo;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.pinot.spi.stream.ConsumerPartitionState;
import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+import org.apache.pinot.spi.stream.PartitionLagState;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
+import org.apache.pinot.spi.stream.StreamMessageMetadata;
import org.apache.pinot.spi.stream.StreamMetadataProvider;
import org.testng.annotations.Test;
@@ -111,6 +114,68 @@ public class KafkaStreamMetadataProviderTest {
}
}
+ @Test
+ public void testGetCurrentPartitionLagStateHandlesInvalidIngestionTime()
+ throws Exception {
+ String topicName = "asset";
+ Consumer<Bytes, Bytes> consumer = mockConsumer(topicName, 1);
+ MOCK_CONSUMER.set(consumer);
+ try {
+ StreamConfig streamConfig = getStreamConfig(topicName);
+ try (KafkaStreamMetadataProvider provider = new
MockKafkaStreamMetadataProvider("client", streamConfig)) {
+ long lastProcessedTimeMs = 1_700_000_100_000L;
+
+ // Record with a valid upstream ingestion time yields a numeric
availability lag.
+ StreamMessageMetadata validMetadata = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .setRecordIngestionTimeMs(lastProcessedTimeMs - 1000L)
+ .build();
+ // Records whose ingestion time is missing/invalid: unset (Builder
default Long.MIN_VALUE), Kafka's
+ // NO_TIMESTAMP (-1), and epoch 0 (the exact boundary of the > 0
guard). These stand in for a topic that is
+ // unreachable/timing out or produces records without a timestamp.
+ StreamMessageMetadata unsetIngestionTime = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .build();
+ StreamMessageMetadata noTimestampIngestionTime = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .setRecordIngestionTimeMs(-1L)
+ .build();
+ StreamMessageMetadata zeroIngestionTime = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .setRecordIngestionTimeMs(0L)
+ .build();
+
+ Map<String, ConsumerPartitionState> stateMap = new HashMap<>();
+ stateMap.put("0", new ConsumerPartitionState("0", new
LongMsgOffset(5), lastProcessedTimeMs,
+ new LongMsgOffset(10), validMetadata));
+ stateMap.put("1", new ConsumerPartitionState("1", new
LongMsgOffset(5), lastProcessedTimeMs,
+ new LongMsgOffset(10), unsetIngestionTime));
+ stateMap.put("2", new ConsumerPartitionState("2", new
LongMsgOffset(5), lastProcessedTimeMs,
+ new LongMsgOffset(10), noTimestampIngestionTime));
+ stateMap.put("3", new ConsumerPartitionState("3", new
LongMsgOffset(5), lastProcessedTimeMs,
+ new LongMsgOffset(10), zeroIngestionTime));
+ // Partition with an unknown upstream offset exercises the offset-lag
fallback.
+ stateMap.put("4", new ConsumerPartitionState("4", new
LongMsgOffset(5), lastProcessedTimeMs,
+ null, validMetadata));
+
+ Map<String, PartitionLagState> lagState =
provider.getCurrentPartitionLagState(stateMap);
+
+ // Offset lag: numeric when both offsets are known, NOT_CALCULATED
when the upstream offset is unavailable.
+ assertEquals(lagState.get("0").getRecordsLag(), "5");
+ assertEquals(lagState.get("4").getRecordsLag(),
PartitionLagState.NOT_CALCULATED);
+ // Availability lag: numeric for a valid ingestion time,
NOT_CALCULATED for every invalid one.
+ // Regression for issue #18836: an invalid ingestion time must not
leak an epoch-sized value
+ // (lastProcessedTimeMs - Long.MIN_VALUE, or lastProcessedTimeMs -
(-1) ~= now).
+ assertEquals(lagState.get("0").getAvailabilityLagMs(), "1000");
+ assertEquals(lagState.get("1").getAvailabilityLagMs(),
PartitionLagState.NOT_CALCULATED);
+ assertEquals(lagState.get("2").getAvailabilityLagMs(),
PartitionLagState.NOT_CALCULATED);
+ assertEquals(lagState.get("3").getAvailabilityLagMs(),
PartitionLagState.NOT_CALCULATED);
+ }
+ } finally {
+ MOCK_CONSUMER.remove();
+ }
+ }
+
private static StreamConfig getStreamConfig(String topicName) {
Map<String, String> streamConfigMap = new HashMap<>();
streamConfigMap.put("streamType", "kafka");
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProvider.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProvider.java
index baa6eeaa610..c3864578920 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProvider.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/main/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProvider.java
@@ -262,10 +262,14 @@ public class KinesisStreamMetadataProvider implements
StreamMetadataProvider {
Map<String, PartitionLagState> perPartitionLag = new HashMap<>();
for (Map.Entry<String, ConsumerPartitionState> entry :
currentPartitionStateMap.entrySet()) {
ConsumerPartitionState partitionState = entry.getValue();
- // Compute record-availability
- String recordAvailabilityLag = "UNKNOWN";
+ // Compute record-availability. Only when both the last-processed
wall-clock time and the record's upstream
+ // ingestion time are valid; otherwise a missing/invalid ingestion time
(e.g. records without a valid
+ // timestamp) would turn the subtraction into an epoch-sized value that
leaks to the metric and UI. Keep the
+ // NOT_CALCULATED sentinel in that case so the lag is reported as
not-calculated rather than a bogus number.
+ String recordAvailabilityLag = PartitionLagState.NOT_CALCULATED;
StreamMessageMetadata lastProcessedMessageMetadata =
partitionState.getLastProcessedRowMetadata();
- if (lastProcessedMessageMetadata != null &&
partitionState.getLastProcessedTimeMs() > 0) {
+ if (lastProcessedMessageMetadata != null &&
partitionState.getLastProcessedTimeMs() > 0
+ && lastProcessedMessageMetadata.getRecordIngestionTimeMs() > 0) {
long availabilityLag =
partitionState.getLastProcessedTimeMs() -
lastProcessedMessageMetadata.getRecordIngestionTimeMs();
recordAvailabilityLag = String.valueOf(availabilityLag);
diff --git
a/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/test/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProviderTest.java
b/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/test/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProviderTest.java
index 59d2b0a4a40..c1e695f3768 100644
---
a/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/test/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProviderTest.java
+++
b/pinot-plugins/pinot-stream-ingestion/pinot-kinesis/src/test/java/org/apache/pinot/plugin/stream/kinesis/KinesisStreamMetadataProviderTest.java
@@ -22,12 +22,16 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import org.apache.pinot.spi.stream.ConsumerPartitionState;
+import org.apache.pinot.spi.stream.LongMsgOffset;
import org.apache.pinot.spi.stream.PartitionGroupConsumer;
import org.apache.pinot.spi.stream.PartitionGroupConsumptionStatus;
import org.apache.pinot.spi.stream.PartitionGroupMetadata;
+import org.apache.pinot.spi.stream.PartitionLagState;
import org.apache.pinot.spi.stream.StreamConfig;
import org.apache.pinot.spi.stream.StreamConfigProperties;
import org.apache.pinot.spi.stream.StreamConsumerFactory;
+import org.apache.pinot.spi.stream.StreamMessageMetadata;
import org.apache.pinot.spi.stream.StreamPartitionMsgOffset;
import org.mockito.ArgumentCaptor;
import org.testng.Assert;
@@ -250,4 +254,48 @@ public class KinesisStreamMetadataProviderTest {
Assert.assertEquals(result.get(0).getPartitionGroupId(), 0);
Assert.assertEquals(partitionGroupMetadataCapture.getValue().getSequenceNumber(),
1);
}
+
+ @Test
+ public void testGetCurrentPartitionLagStateHandlesInvalidIngestionTime() {
+ long lastProcessedTimeMs = 1_700_000_100_000L;
+
+ // Shard with a valid upstream ingestion time yields a numeric
availability lag.
+ StreamMessageMetadata validMetadata = new StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .setRecordIngestionTimeMs(lastProcessedTimeMs - 1000L)
+ .build();
+ // Shards whose ingestion time is missing/invalid: unset (Builder default
Long.MIN_VALUE), NO_TIMESTAMP (-1),
+ // and epoch 0 (the exact boundary of the > 0 guard). These stand in for a
stream that is unreachable/timing
+ // out or produces records without a timestamp.
+ StreamMessageMetadata unsetIngestionTime = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .build();
+ StreamMessageMetadata noTimestampIngestionTime = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .setRecordIngestionTimeMs(-1L)
+ .build();
+ StreamMessageMetadata zeroIngestionTime = new
StreamMessageMetadata.Builder()
+ .setOffset(new LongMsgOffset(5), new LongMsgOffset(6))
+ .setRecordIngestionTimeMs(0L)
+ .build();
+
+ Map<String, ConsumerPartitionState> stateMap = new HashMap<>();
+ stateMap.put("0", new ConsumerPartitionState("0", new LongMsgOffset(5),
lastProcessedTimeMs,
+ new LongMsgOffset(10), validMetadata));
+ stateMap.put("1", new ConsumerPartitionState("1", new LongMsgOffset(5),
lastProcessedTimeMs,
+ new LongMsgOffset(10), unsetIngestionTime));
+ stateMap.put("2", new ConsumerPartitionState("2", new LongMsgOffset(5),
lastProcessedTimeMs,
+ new LongMsgOffset(10), noTimestampIngestionTime));
+ stateMap.put("3", new ConsumerPartitionState("3", new LongMsgOffset(5),
lastProcessedTimeMs,
+ new LongMsgOffset(10), zeroIngestionTime));
+
+ Map<String, PartitionLagState> lagState =
_kinesisStreamMetadataProvider.getCurrentPartitionLagState(stateMap);
+
+ Assert.assertEquals(lagState.get("0").getAvailabilityLagMs(), "1000");
+ // Regression for issue #18836: an invalid ingestion time must report the
NOT_CALCULATED sentinel instead of an
+ // epoch-sized value (lastProcessedTimeMs - Long.MIN_VALUE, or
lastProcessedTimeMs - (-1) ~= now).
+ Assert.assertEquals(lagState.get("1").getAvailabilityLagMs(),
PartitionLagState.NOT_CALCULATED);
+ Assert.assertEquals(lagState.get("2").getAvailabilityLagMs(),
PartitionLagState.NOT_CALCULATED);
+ Assert.assertEquals(lagState.get("3").getAvailabilityLagMs(),
PartitionLagState.NOT_CALCULATED);
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
