You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "wuguowei1994 (via GitHub)" <gi...@apache.org> on 2025/11/18 06:21:16 UTC
[PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
wuguowei1994 opened a new pull request, #18750:
URL: https://github.com/apache/druid/pull/18750
# Motivation
We operate a Druid deployment with more than 500 nodes.
In real-time ingestion scenarios, a monitoring process queries the cluster every minute to retrieve the `ingest/kafka/partitionLag` metric. If the lag remains unhealthy for more than five minutes, alerts are triggered.
In our production environment, this metric periodically becomes **negative**, even when the cluster is fully healthy. These false alerts create unnecessary operational load and frequently wake the on-call team during off-hours. On the other hand, we cannot suppress negative-lag alerts entirely, since in some situations negative lag can indicate real ingestion problems.
For a large-scale, 24×7 real-time ingestion pipeline, **accurate and consistent lag metrics are essential to avoid unnecessary nighttime wake-ups while still ensuring that real issues are detected promptly**.
---
# Problem Description
<img width="1255" height="839" alt="negative_lag" src="https://github.com/user-attachments/assets/3b193544-984d-4d62-9387-06d0100b79f7" />
In the current implementation, the Druid supervisor maintains two volatile data structures:
* The latest Kafka `end_offset` for each partition
* The latest task-reported `current_offset` for each partition
The supervisor periodically updates these values (every 30 seconds):
1. Querying all tasks in parallel to update `current_offset`.
This step waits for all HTTP requests to complete and each request has a timeout of two minutes.
2. Querying Kafka cluster to refresh `end_offset`.
On the other hand, a separate periodic task (every minute) computes:
```
lag = end_offset - current_offset
```
Because the two updates are not atomic, intermediate inconsistent states may occur.
### Intermediate State Leading to Negative Lag
If one task becomes heavily loaded or experiences other delays during Step 1, it may take significantly longer to return its offset. In this situation, the supervisor continues waiting for that slow task while the other tasks have already responded.
During this waiting period:
* Many `current_offset` values already have been updated to new values.
* The `end_offset` values remain stale because Step 2 has not executed yet.
If a monitoring request arrives in this intermediate window, the supervisor computes lag using:
* **Partially updated `current_offset`**
* **Stale `end_offset`**
This produces negative lag values.
This issue repeats as long as at least one task remains slow. Large clusters with many partitions and many Kafka-indexing tasks are more likely to experience this scenario.
---
# Example Scenario
1. Initial state: `end_offset = 10000`, `current_offset = 0`.
2. After consumption: latest Kafka `end_offset = 30000`, and all tasks have consumed up to `20000`.
3. During Step 1, 49 tasks respond quickly, and their `current_offset` is updated to `20000`.
One task is slow, causing Step 1 to remain in the awaiting state.
4. The in-memory `end_offset` stays at the old value `10000`.
5. If a metric query occurs at this point, the supervisor calculates:
```
10000 - 20000 = -10000
```
6. Because the periodic update logic repeats, this situation can persist across multiple cycles.
---
# Proposed Changes
Replace the two volatile structures storing `current_offset` and `end_offset` with `AtomicReference` containers that hold both values as a single immutable state object. The supervisor will update these references as atomic units, ensuring that lag computation always observes a consistent snapshot.
This eliminates inconsistent intermediate states and prevents negative lag due to partial updates.
---
# Rationale
* Ensures consistent reads between related fields.
* Avoids breaking existing APIs or supervisor behavior.
* The change is localized to offset bookkeeping logic.
* No behavioral changes other than removing negative lag caused by inconsistent state.
---
# Operational Impact
* No configuration changes required.
* No backward incompatibility.
* Improved accuracy of Kafka lag metrics in large clusters.
* Reduces false alerts in monitoring systems.
---
# Test Plan
* This change does not add new feature. We only need to make sure existing tests still pass.
* All current tests pass successfully.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3555380405
I checked out the latest master branch and ran the tests locally — the same MSQInsertTest and MSQReplaceTest failures (timestamp off by 2 days and shuffle counter length mismatch) are still present there. So it looks like these failures are not introduced by my changes.
Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2621676522
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshotTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetSnapshotTest
+{
+ @Test
+ public void testOffsetSnapshot_emptyInputReturnsEmptyMap()
+ {
+ OffsetSnapshot<String, Long> snapshot = OffsetSnapshot.of(
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ );
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullInputsReturnEmptyMaps()
+ {
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, null);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullCurrentOffsetsReturnsEmptyCurrentMap()
+ {
+ Map<Integer, Long> endOffsets = ImmutableMap.of(0, 100L, 1, 200L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, endOffsets);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(endOffsets, snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullEndOffsetsReturnsEmptyEndMap()
+ {
+ Map<Integer, Long> currentOffsets = ImmutableMap.of(0, 50L, 1, 150L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(currentOffsets, null);
+
+ Assert.assertEquals(currentOffsets, snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_copiesInputMapsAndReturnsImmutableCopies()
+ {
+ Map<String, String> current = new HashMap<>();
+ current.put("p0", "100");
+ current.put("p1", "200");
+
+ Map<String, String> end = new HashMap<>();
+ end.put("p0", "150");
+ end.put("p2", "300");
+
+ OffsetSnapshot<String, String> snapshot = OffsetSnapshot.of(current, end);
+
+ Assert.assertEquals(ImmutableMap.of("p0", "100", "p1", "200"), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(ImmutableMap.of("p0", "150", "p2", "300"), snapshot.getLatestOffsetsFromStream());
+
+ // Returned maps must be immutable
Review Comment:
i think strict compiler is complaining about this, also, no need to test ImmutableMap.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3581028713
@clintropolis
Thank you very much for your sincere reply. I've thought about your code suggestions, made the improvements, and finished the changes. Could you please review them again?
In fact, our company is also a paid user of the commercial version of Imply, and we spend quite a lot of money on it every year.
As an engineer working on the company's OLAP platform, I've been involved in the Druid project for almost three years. I've studied a large amount of Druid's source code and debugged many production issues. In the past, my code changes were only submitted internally within the company. Now, since our U.S.-based company is about to wind down its operations in China, I'm also facing layoffs soon and don't have much work these days. That's why I finally have time to clean up my previous code and submit it upstream.
Losing my job has made me a bit anxious lately, and the recent bug-fix progress has been quite slow, so I ended up venting a little. I hope you don't mind.
Lastly, I really want to take advantage of this free time to get all my code submitted as soon as possible. Thank you so much!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3631604706
@kfaraz @cecemei
Hey, could you check out the code? Suggestions welcome! 😄
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3555338406
@abhishekrb19 I checked the details of the failing UT job. The failures in MSQInsertTest and MSQReplaceTest (timestamp off by exactly 2 days and shuffle counter length mismatch) don’t seem related to my Kafka offset changes at all. It looks a lot like the classic CI timezone/flaky issue that pops up from time to time.
Could someone help take a quick look? Thanks!
https://github.com/apache/druid/actions/runs/19487515271/job/55886862437
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3555577863
@wuguowei1994 could you sync the latest master to your branch? I think [this](https://github.com/apache/druid/pull/18753) patch might fix the test failures.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3589379827
Could any kind people help me merge it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2563010364
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -294,13 +295,16 @@ protected Map<KafkaTopicPartition, Long> getPartitionTimeLag()
// suppress use of CollectionUtils.mapValues() since the valueMapper function is dependent on map key here
@SuppressWarnings("SSBasedInspection")
// Used while calculating cummulative lag for entire stream
- private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(Map<KafkaTopicPartition, Long> currentOffsets)
+ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences()
{
- if (latestSequenceFromStream == null) {
+ Map<KafkaTopicPartition, Long> currentOffsets = offsetSnapshotRef.get().getCurrentOffsets();
+ Map<KafkaTopicPartition, Long> endOffsets = offsetSnapshotRef.get().getEndOffsets();
Review Comment:
seems like we should break out `offsetSnapshotRef.get()` so that these two offsets maps are from the same `OffsetSnapshot`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3550391926
@FrankChen021 This is my first time submitting a PR to Druid. If I’d like to speed up the review, who would be the right person to @?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2589875135
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
Review Comment:
Please use the field names `latestOffsetsFromStream`, `highestIngestedOffsets` to avoid confusion. `currentOffsets` and `endOffsets` are too similar.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
+ this.endOffsets = toImmutableOffsetMap(endOffsets);
+ }
+
+ private ImmutableMap<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null || input.isEmpty()) {
+ return ImmutableMap.of();
+ }
+
+ return input.entrySet().stream()
+ .filter(e -> e.getValue() != null)
+ .collect(ImmutableMap.toImmutableMap(
+ Map.Entry::getKey,
+ Map.Entry::getValue
+ ));
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getCurrentOffsets()
+ {
+ return currentOffsets;
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getEndOffsets()
+ {
+ return endOffsets;
+ }
+ }
+
+ protected final AtomicReference<OffsetSnapshot<PartitionIdType, SequenceOffsetType>> offsetSnapshotRef =
+ new AtomicReference<>(new OffsetSnapshot<>(Collections.emptyMap(), Collections.emptyMap()));
Review Comment:
Nit: for brevity, might be nice to have a static `of` method in `OffsetSnapshot`.
for example:
```suggestion
new AtomicReference<>(OffsetSnapshot.of(Map.of(), Map.of()));
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
Review Comment:
Please put this class in a separate file of its own since the supervisor class is already too bloated.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -172,15 +172,15 @@ protected SeekableStreamSupervisorReportPayload<KafkaTopicPartition, Long> creat
)
{
KafkaSupervisorIOConfig ioConfig = spec.getIoConfig();
- Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences(getHighestCurrentOffsets());
+ Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences();
return new KafkaSupervisorReportPayload(
spec.getId(),
spec.getDataSchema().getDataSource(),
ioConfig.getStream(),
numPartitions,
ioConfig.getReplicas(),
ioConfig.getTaskDuration().getMillis() / 1000,
- includeOffsets ? latestSequenceFromStream : null,
+ includeOffsets ? offsetSnapshotRef.get().getEndOffsets() : null,
Review Comment:
+1
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
Review Comment:
+1 to the javadocs
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
+ this.endOffsets = toImmutableOffsetMap(endOffsets);
+ }
+
+ private ImmutableMap<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null || input.isEmpty()) {
+ return ImmutableMap.of();
+ }
+
+ return input.entrySet().stream()
+ .filter(e -> e.getValue() != null)
+ .collect(ImmutableMap.toImmutableMap(
+ Map.Entry::getKey,
+ Map.Entry::getValue
+ ));
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getCurrentOffsets()
+ {
+ return currentOffsets;
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getEndOffsets()
+ {
+ return endOffsets;
+ }
+ }
+
+ protected final AtomicReference<OffsetSnapshot<PartitionIdType, SequenceOffsetType>> offsetSnapshotRef =
Review Comment:
Let's move this field to `KafkaSupervisor` and use it only for Kafka for the time being.
If it works fine there, we will adopt it later for the other supervisors too.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -454,6 +461,12 @@ private void updatePartitionTimeAndRecordLagFromStream()
e -> e.getValue() - lastIngestedTimestamps.get(e.getKey())
)
);
+
+ OffsetSnapshot<KafkaTopicPartition, Long> snapshot = new OffsetSnapshot<>(
+ highestCurrentOffsets,
+ latestSequenceFromStream
+ );
+ offsetSnapshotRef.set(snapshot);
Review Comment:
Nit: can be in a single statement
```suggestion
offsetSnapshotRef.set(
OffsetSnapshot.of(highestCurrentOffsets, latestSequenceFromStream)
);
```
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -526,6 +539,12 @@ protected void updatePartitionLagFromStream()
latestSequenceFromStream =
partitions.stream().collect(Collectors.toMap(StreamPartition::getPartitionId, recordSupplier::getPosition));
+
+ OffsetSnapshot<KafkaTopicPartition, Long> snapshot = new OffsetSnapshot<>(
Review Comment:
+1
After the `latestSequenceFromStream` has been fetched, another thread could have updated the metadata store with the latest offsets which are returned from the `getHighestCurrentOffsets()`.
If you just make the change as @cecemei suggests, I think we could ensure positive lag in this case as well.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -172,15 +172,15 @@ protected SeekableStreamSupervisorReportPayload<KafkaTopicPartition, Long> creat
)
{
KafkaSupervisorIOConfig ioConfig = spec.getIoConfig();
- Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences(getHighestCurrentOffsets());
+ Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences();
return new KafkaSupervisorReportPayload(
spec.getId(),
spec.getDataSchema().getDataSource(),
ioConfig.getStream(),
numPartitions,
ioConfig.getReplicas(),
ioConfig.getTaskDuration().getMillis() / 1000,
- includeOffsets ? latestSequenceFromStream : null,
+ includeOffsets ? offsetSnapshotRef.get().getEndOffsets() : null,
Review Comment:
re-use method instead
```suggestion
includeOffsets ? getLatestSequencesFromStream() : null,
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3690666658
Could anyone take a look, please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3979729624
Our company has an urgent project that needs to go live within the next two weeks. I plan to refine and complete this round of changes after the 15th of this month. Thank you!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2621822595
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshotTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetSnapshotTest
+{
+ @Test
+ public void testOffsetSnapshot_emptyInputReturnsEmptyMap()
+ {
+ OffsetSnapshot<String, Long> snapshot = OffsetSnapshot.of(
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ );
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullInputsReturnEmptyMaps()
+ {
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, null);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullCurrentOffsetsReturnsEmptyCurrentMap()
+ {
+ Map<Integer, Long> endOffsets = ImmutableMap.of(0, 100L, 1, 200L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, endOffsets);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(endOffsets, snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullEndOffsetsReturnsEmptyEndMap()
+ {
+ Map<Integer, Long> currentOffsets = ImmutableMap.of(0, 50L, 1, 150L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(currentOffsets, null);
+
+ Assert.assertEquals(currentOffsets, snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_copiesInputMapsAndReturnsImmutableCopies()
+ {
+ Map<String, String> current = new HashMap<>();
+ current.put("p0", "100");
+ current.put("p1", "200");
+
+ Map<String, String> end = new HashMap<>();
+ end.put("p0", "150");
+ end.put("p2", "300");
+
+ OffsetSnapshot<String, String> snapshot = OffsetSnapshot.of(current, end);
+
+ Assert.assertEquals(ImmutableMap.of("p0", "100", "p1", "200"), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(ImmutableMap.of("p0", "150", "p2", "300"), snapshot.getLatestOffsetsFromStream());
+
+ // Returned maps must be immutable
+ Assert.assertThrows(
+ UnsupportedOperationException.class,
+ () -> snapshot.getHighestIngestedOffsets().put("x", "x")
Review Comment:
fixed
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshotTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetSnapshotTest
+{
+ @Test
+ public void testOffsetSnapshot_emptyInputReturnsEmptyMap()
+ {
+ OffsetSnapshot<String, Long> snapshot = OffsetSnapshot.of(
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ );
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullInputsReturnEmptyMaps()
+ {
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, null);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullCurrentOffsetsReturnsEmptyCurrentMap()
+ {
+ Map<Integer, Long> endOffsets = ImmutableMap.of(0, 100L, 1, 200L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, endOffsets);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(endOffsets, snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullEndOffsetsReturnsEmptyEndMap()
+ {
+ Map<Integer, Long> currentOffsets = ImmutableMap.of(0, 50L, 1, 150L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(currentOffsets, null);
+
+ Assert.assertEquals(currentOffsets, snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_copiesInputMapsAndReturnsImmutableCopies()
+ {
+ Map<String, String> current = new HashMap<>();
+ current.put("p0", "100");
+ current.put("p1", "200");
+
+ Map<String, String> end = new HashMap<>();
+ end.put("p0", "150");
+ end.put("p2", "300");
+
+ OffsetSnapshot<String, String> snapshot = OffsetSnapshot.of(current, end);
+
+ Assert.assertEquals(ImmutableMap.of("p0", "100", "p1", "200"), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(ImmutableMap.of("p0", "150", "p2", "300"), snapshot.getLatestOffsetsFromStream());
+
+ // Returned maps must be immutable
+ Assert.assertThrows(
+ UnsupportedOperationException.class,
+ () -> snapshot.getHighestIngestedOffsets().put("x", "x")
+ );
+ Assert.assertThrows(
+ UnsupportedOperationException.class,
+ () -> snapshot.getLatestOffsetsFromStream().put("x", "x")
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3707893237
Take a look? Jan 4th
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3578116514
There are two failing checks:
`Caused by: java.net.ConnectException: Connection refused: localhost/127.0.0.1:30400`
It appears that the issue is not caused by my code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3550310563
@FrankChen021
https://github.com/apache/druid/actions/runs/19461511000/job/55719076867?pr=18750
Could you help me re-run the failing UT job? The error doesn’t look like it was caused by my code.
```
Installed distributions
Trying to resolve the latest version from remote
Error: error code: 500
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3654774396
@cecemei Sorry, I’ve been dealing with some personal matters recently, so my reply is a bit late. I’ve gone through all your reviews— they’re excellent, thank you! It was also a great learning opportunity for me.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3578456965
```
ERROR [main] org.apache.druid.testing.utils.DruidClusterAdminClient
- Error while waiting for [http://localhost:30400] to be ready
java.util.concurrent.ExecutionException: org.jboss.netty.channel.ChannelException: Faulty channel in resource pool
```
@clintropolis Thanks for helping me rerun the jobs. Before the rerun there were two failed tasks, and now there’s only one left. The error still looks like an environment issue… So sad.
**This experience submitting code shows me that the project is no longer very active. Many people report issues in the Druid Slack, and almost no one responds. No wonder so many companies are migrating from Druid to ClickHouse.**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "clintropolis (via GitHub)" <gi...@apache.org>.
clintropolis commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3579070636
> @clintropolis Thanks for rerunning the jobs. Before the rerun there were two failed tasks, and now there’s only one left. The error still looks like an environment issue… really frustrating.
Yea, we are in the middle of a bit of a migration/overhaul of our integration test framework and processes, hopefully this will be more well behaved in the future, since part of the reason for this change is to address flakiness as well as make it much easier to write and debug these tests.
> This whole code submission experience also shows how inactive the project has become. A lot of people report issues in the Druid Slack, but almost no one responds anymore. It’s no surprise so many companies are migrating from Druid to ClickHouse.
Apologies for the perception - while I can assure you that there are quite a lot of active and interesting projects happening and that Druid is still very much being actively developed, that doesn't fix your experience here or the optics in slack. The unfortunate matter is that there are always a lot more things to do than people to do them, and while we try our best, sometimes things are a bit slow to get a committers attention. All that said, thanks again for the contribution and trying to make things better, and I do hear you - i'll see if maybe I can nudge some of the other committers to be a bit more responsive and try to do the same myself.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2621819211
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshot.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+
+import javax.annotation.Nullable;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Immutable snapshot containing a consistent pair of offset maps: the highest ingested offsets
+ * reported by tasks and the latest end offsets fetched from the underlying stream.
+ *
+ * <p>
+ * The supervisor fetches task-reported ingested offsets first, then fetches end offsets from the stream.
+ * Because these two values are captured at slightly different instants, the reported lag
+ * (latestOffsetsFromStream - highestIngestedOffsets) may be slightly larger than the actual lag at any
+ * precise moment.
+ *
+ * <p>
+ * By publishing both maps together as a single atomic snapshot (using {@link java.util.concurrent.atomic.AtomicReference}),
+ * readers (such as lag metrics and supervisor status) always observe a coherent and consistent view.
+ * This produces stable and monotonic lag trends, avoiding artifacts like temporary negative lags.
+ *
+ * <p>
+ * This class is generic and can be reused by all seekable-stream supervisors (Kafka, Kinesis, etc.).
+ */
+public final class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+{
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> highestIngestedOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> latestOffsetsFromStream;
+
+ private OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> highestIngestedOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> latestOffsetsFromStream
+ )
+ {
+ this.highestIngestedOffsets = toImmutableOffsetMap(highestIngestedOffsets);
+ this.latestOffsetsFromStream = toImmutableOffsetMap(latestOffsetsFromStream);
+ }
+
+ public static <PartitionIdType, SequenceOffsetType> OffsetSnapshot<PartitionIdType, SequenceOffsetType> of(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ return new OffsetSnapshot<>(currentOffsets, endOffsets);
+ }
+
+ private ImmutableMap<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null) {
+ return ImmutableMap.of();
+ }
+
+ return ImmutableMap.copyOf(Maps.filterValues(input, Objects::nonNull));
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getHighestIngestedOffsets()
Review Comment:
false positive, field is ImmutableMap
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshot.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+
+import javax.annotation.Nullable;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Immutable snapshot containing a consistent pair of offset maps: the highest ingested offsets
+ * reported by tasks and the latest end offsets fetched from the underlying stream.
+ *
+ * <p>
+ * The supervisor fetches task-reported ingested offsets first, then fetches end offsets from the stream.
+ * Because these two values are captured at slightly different instants, the reported lag
+ * (latestOffsetsFromStream - highestIngestedOffsets) may be slightly larger than the actual lag at any
+ * precise moment.
+ *
+ * <p>
+ * By publishing both maps together as a single atomic snapshot (using {@link java.util.concurrent.atomic.AtomicReference}),
+ * readers (such as lag metrics and supervisor status) always observe a coherent and consistent view.
+ * This produces stable and monotonic lag trends, avoiding artifacts like temporary negative lags.
+ *
+ * <p>
+ * This class is generic and can be reused by all seekable-stream supervisors (Kafka, Kinesis, etc.).
+ */
+public final class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+{
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> highestIngestedOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> latestOffsetsFromStream;
+
+ private OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> highestIngestedOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> latestOffsetsFromStream
+ )
+ {
+ this.highestIngestedOffsets = toImmutableOffsetMap(highestIngestedOffsets);
+ this.latestOffsetsFromStream = toImmutableOffsetMap(latestOffsetsFromStream);
+ }
+
+ public static <PartitionIdType, SequenceOffsetType> OffsetSnapshot<PartitionIdType, SequenceOffsetType> of(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ return new OffsetSnapshot<>(currentOffsets, endOffsets);
+ }
+
+ private ImmutableMap<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null) {
+ return ImmutableMap.of();
+ }
+
+ return ImmutableMap.copyOf(Maps.filterValues(input, Objects::nonNull));
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getHighestIngestedOffsets()
+ {
+ return highestIngestedOffsets;
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getLatestOffsetsFromStream()
Review Comment:
false positive, field is ImmutableMap
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3658904077
@cecemei Fixed the issue with UT and merged the latest changes from master. Thank you very much for your support throughout the entire process!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3752500603
@clintropolis @kfaraz Take a look, please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3550579832
@wuguowei1994 , thanks for creating a PR in Apache Druid! One of the contributors will take a look at the changes soon.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3615163249
> In fact, the negative lag could even be a feature to identify if some tasks are particularly slow in returning their offsets. 😛 , and we could probably have alerts set up if the negative lag goes below a specific threshold.
>
@kfaraz
Thanks for the clarification — that makes sense. In our case, though, we’ve noticed that negative lag in our large cluster can sometimes persist for over five minutes.
We’ve talked about this internally, and if it only happens occasionally (for example, under a minute), adjusting the alert thresholds would absolutely work for us. But when it lasts longer, it tends to indicate something worth investigating.
**We’ve also seen a few situations where negative lag actually pointed to issues in the upstream Kafka cluster, so that’s part of why we’re a bit cautious here. If we keep the current Druid behavior and treat negative lag as normal consumption, there’s a chance we might overlook real problems.**
So overall, having clear and reliable metrics to signal the health of the cluster would be really helpful for us.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2604952312
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -294,44 +298,48 @@ protected Map<KafkaTopicPartition, Long> getPartitionTimeLag()
// suppress use of CollectionUtils.mapValues() since the valueMapper function is dependent on map key here
@SuppressWarnings("SSBasedInspection")
// Used while calculating cummulative lag for entire stream
- private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(Map<KafkaTopicPartition, Long> currentOffsets)
+ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences()
{
- if (latestSequenceFromStream == null) {
+ Map<KafkaTopicPartition, Long> highestIngestedOffsets = getHighestIngestedOffsets();
+ Map<KafkaTopicPartition, Long> latestSequencesFromStream = getLatestSequencesFromStream();
+
+ if (latestSequencesFromStream.isEmpty()) {
return Collections.emptyMap();
}
- return latestSequenceFromStream
- .entrySet()
- .stream()
- .collect(
- Collectors.toMap(
- Entry::getKey,
- e -> e.getValue() != null
- ? e.getValue() - Optional.ofNullable(currentOffsets.get(e.getKey())).orElse(0L)
- : 0
- )
- );
+ return latestSequencesFromStream.entrySet()
+ .stream()
+ .collect(
+ Collectors.toMap(
+ Entry::getKey,
+ e ->
+ e.getValue() - highestIngestedOffsets.getOrDefault(e.getKey(), 0L)
+ )
+ );
}
+ // This function is defined and called by the parent class to compute the lag for specific partitions.
Review Comment:
a few notes:
- please use javadoc style documentation for method.
- worth noting that the offset in `currentOffsets` is ignored since we adopt the `OffsetSnapshot` approach. this means the `currentOffsets` is no longer used in the lag calculation in the `TaskReportData`.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -294,44 +298,48 @@ protected Map<KafkaTopicPartition, Long> getPartitionTimeLag()
// suppress use of CollectionUtils.mapValues() since the valueMapper function is dependent on map key here
@SuppressWarnings("SSBasedInspection")
// Used while calculating cummulative lag for entire stream
- private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(Map<KafkaTopicPartition, Long> currentOffsets)
+ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences()
{
- if (latestSequenceFromStream == null) {
+ Map<KafkaTopicPartition, Long> highestIngestedOffsets = getHighestIngestedOffsets();
+ Map<KafkaTopicPartition, Long> latestSequencesFromStream = getLatestSequencesFromStream();
+
+ if (latestSequencesFromStream.isEmpty()) {
return Collections.emptyMap();
}
- return latestSequenceFromStream
- .entrySet()
- .stream()
- .collect(
- Collectors.toMap(
- Entry::getKey,
- e -> e.getValue() != null
- ? e.getValue() - Optional.ofNullable(currentOffsets.get(e.getKey())).orElse(0L)
- : 0
- )
- );
+ return latestSequencesFromStream.entrySet()
+ .stream()
+ .collect(
+ Collectors.toMap(
+ Entry::getKey,
+ e ->
+ e.getValue() - highestIngestedOffsets.getOrDefault(e.getKey(), 0L)
+ )
+ );
}
+ // This function is defined and called by the parent class to compute the lag for specific partitions.
+ // The `currentOffsets` parameter is provided by the parent class and indicates the partitions to query.
+ // Note: This function differs from `getRecordLagPerPartitionInLatestSequences()`:
+ // `getRecordLagPerPartitionInLatestSequences()` queries lag for all partitions,
+ // whereas `getRecordLagPerPartition()` queries lag only for the specified partitions.
@Override
protected Map<KafkaTopicPartition, Long> getRecordLagPerPartition(Map<KafkaTopicPartition, Long> currentOffsets)
{
- if (latestSequenceFromStream == null || currentOffsets == null) {
+ Map<KafkaTopicPartition, Long> latestSequencesFromStream = getLatestSequencesFromStream();
Review Comment:
same here, dont break atomicity:
```
OffsetSnapshot<KafkaTopicPartition, Long> offsetSnapshot = offsetSnapshotRef.get();
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshot.java:
##########
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+
+import javax.annotation.Nullable;
+import java.util.Map;
+import java.util.Objects;
+
+/**
Review Comment:
A few notes:
- use `<p>` as paragraph tag
- there's no need to include too much details on previous behavior or link the pr usually, as the documentation should be forward-looking. Optionally you could add a release note section in your PR to document this change. For the javadoc, maybe just could say like:
>The supervisor fetches task-reported ingested offsets first, then fetches end offsets from the stream. Because these two values are captured at different instants, the reported delay (latestOffsetsFromStream - highestIngestedOffsets) may be slightly larger than the actual delay at any precise moment. However, publishing both maps together as a single atomic snapshot ensures that readers always observe a consistent view, producing stable and monotonic delay trends.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -294,44 +298,48 @@ protected Map<KafkaTopicPartition, Long> getPartitionTimeLag()
// suppress use of CollectionUtils.mapValues() since the valueMapper function is dependent on map key here
@SuppressWarnings("SSBasedInspection")
// Used while calculating cummulative lag for entire stream
- private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(Map<KafkaTopicPartition, Long> currentOffsets)
+ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences()
{
- if (latestSequenceFromStream == null) {
+ Map<KafkaTopicPartition, Long> highestIngestedOffsets = getHighestIngestedOffsets();
+ Map<KafkaTopicPartition, Long> latestSequencesFromStream = getLatestSequencesFromStream();
Review Comment:
dont call for the offsets separately, it's breaking the atomicity, instead:
pass in `OffsetSnapshot<KafkaTopicPartition, Long> offsetSnapshot` as param.
```
offsetSnapshot.getLatestOffsetsFromStream()
.entrySet()
.stream()
.collect(Collectors.toMap(
Entry::getKey,
e -> e.getValue() - offsetSnapshot.getHighestIngestedOffsets().getOrDefault(e.getKey(), 0L)
));
```
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -294,44 +298,48 @@ protected Map<KafkaTopicPartition, Long> getPartitionTimeLag()
// suppress use of CollectionUtils.mapValues() since the valueMapper function is dependent on map key here
@SuppressWarnings("SSBasedInspection")
// Used while calculating cummulative lag for entire stream
- private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(Map<KafkaTopicPartition, Long> currentOffsets)
+ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences()
{
- if (latestSequenceFromStream == null) {
+ Map<KafkaTopicPartition, Long> highestIngestedOffsets = getHighestIngestedOffsets();
+ Map<KafkaTopicPartition, Long> latestSequencesFromStream = getLatestSequencesFromStream();
+
+ if (latestSequencesFromStream.isEmpty()) {
return Collections.emptyMap();
}
- return latestSequenceFromStream
- .entrySet()
- .stream()
- .collect(
- Collectors.toMap(
- Entry::getKey,
- e -> e.getValue() != null
- ? e.getValue() - Optional.ofNullable(currentOffsets.get(e.getKey())).orElse(0L)
- : 0
- )
- );
+ return latestSequencesFromStream.entrySet()
+ .stream()
+ .collect(
+ Collectors.toMap(
+ Entry::getKey,
+ e ->
+ e.getValue() - highestIngestedOffsets.getOrDefault(e.getKey(), 0L)
+ )
+ );
}
+ // This function is defined and called by the parent class to compute the lag for specific partitions.
+ // The `currentOffsets` parameter is provided by the parent class and indicates the partitions to query.
+ // Note: This function differs from `getRecordLagPerPartitionInLatestSequences()`:
+ // `getRecordLagPerPartitionInLatestSequences()` queries lag for all partitions,
+ // whereas `getRecordLagPerPartition()` queries lag only for the specified partitions.
@Override
protected Map<KafkaTopicPartition, Long> getRecordLagPerPartition(Map<KafkaTopicPartition, Long> currentOffsets)
{
- if (latestSequenceFromStream == null || currentOffsets == null) {
+ Map<KafkaTopicPartition, Long> latestSequencesFromStream = getLatestSequencesFromStream();
+ Map<KafkaTopicPartition, Long> highestIngestedOffsets = getHighestIngestedOffsets();
+
+ if (latestSequencesFromStream.isEmpty() || highestIngestedOffsets.isEmpty() || currentOffsets == null) {
Review Comment:
`highestIngestedOffsets` can be empty right? meaning ingestion hasnt start so offset default to 0.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -535,10 +549,24 @@ protected void updatePartitionLagFromStream()
}
}
+ private void updateOffsetSnapshot(
+ Map<KafkaTopicPartition, Long> highestIngestedOffsets,
+ Map<KafkaTopicPartition, Long> latestOffsetsFromStream
+ )
+ {
+ offsetSnapshotRef.set(
+ OffsetSnapshot.of(highestIngestedOffsets, latestOffsetsFromStream));
+ }
+
@Override
protected Map<KafkaTopicPartition, Long> getLatestSequencesFromStream()
{
- return latestSequenceFromStream != null ? latestSequenceFromStream : new HashMap<>();
+ return offsetSnapshotRef.get().getLatestOffsetsFromStream();
+ }
+
+ private Map<KafkaTopicPartition, Long> getHighestIngestedOffsets()
Review Comment:
the method `getLatestSequencesFromStream` and `getHighestIngestedOffsets` breaks the atomicity of the `OffsetSnapshot` class. imagine, you called for `getHighestIngestedOffsets` then the `offsetSnapshotRef` got updated, then you call for `getLatestSequencesFromStream`, now you're literally comparing offsets at different times!
it's okie to keep `getLatestSequencesFromStream` for compatibility, but please try to not use it in kafkaSupervisor, and in all places, just call `offsetSnapshotRef.get()` for the offset directly.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -172,15 +175,15 @@ protected SeekableStreamSupervisorReportPayload<KafkaTopicPartition, Long> creat
)
{
KafkaSupervisorIOConfig ioConfig = spec.getIoConfig();
- Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences(getHighestCurrentOffsets());
+ Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences();
return new KafkaSupervisorReportPayload(
spec.getId(),
spec.getDataSchema().getDataSource(),
ioConfig.getStream(),
numPartitions,
ioConfig.getReplicas(),
ioConfig.getTaskDuration().getMillis() / 1000,
- includeOffsets ? latestSequenceFromStream : null,
+ includeOffsets ? getLatestSequencesFromStream() : null,
Review Comment:
this sort of breaks the atomicity, please make sure the same `OffsetSnapshot<KafkaTopicPartition, Long>` is used, i.e. pass `OffsetSnapshot<KafkaTopicPartition, Long>` as param to `getRecordLagPerPartitionInLatestSequences`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3669895466
@kfaraz @clintropolis Take a look, please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3693578452
Take a look ? Dec 27
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3688335510
@kfaraz @clintropolis Take some time to review it, please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3555420386
I tested it locally using the latest code from the master branch, and the two unit tests, **MSQInsertTest** and **MSQReplaceTest**, still have issues. This proves that the failures were not caused by my changes.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3938905111
Sure, no hurry
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2564701652
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -294,13 +295,16 @@ protected Map<KafkaTopicPartition, Long> getPartitionTimeLag()
// suppress use of CollectionUtils.mapValues() since the valueMapper function is dependent on map key here
@SuppressWarnings("SSBasedInspection")
// Used while calculating cummulative lag for entire stream
- private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(Map<KafkaTopicPartition, Long> currentOffsets)
+ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences()
{
- if (latestSequenceFromStream == null) {
+ Map<KafkaTopicPartition, Long> currentOffsets = offsetSnapshotRef.get().getCurrentOffsets();
+ Map<KafkaTopicPartition, Long> endOffsets = offsetSnapshotRef.get().getEndOffsets();
Review Comment:
Thanks!
Let me fix it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3612011993
@cecemei I've gone through your comments thoroughly — they're great! Give me a bit of time to improve the code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3691987895
Could someone kind please review the code?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3581101975
@clintropolis If your team is too busy, feel free to assign some issues to me. I’ve got some free time lately anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2587006387
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -172,15 +172,15 @@ protected SeekableStreamSupervisorReportPayload<KafkaTopicPartition, Long> creat
)
{
KafkaSupervisorIOConfig ioConfig = spec.getIoConfig();
- Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences(getHighestCurrentOffsets());
+ Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences();
Review Comment:
maybe pass `OffsetSnapshot` as param so that what's used in the lag calculation is consistent with line 183?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
Review Comment:
It’s not clear to me under what circumstances the inputs could be null. Could you document the scenarios where this might occur?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
+ this.endOffsets = toImmutableOffsetMap(endOffsets);
+ }
+
+ private ImmutableMap<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null || input.isEmpty()) {
+ return ImmutableMap.of();
+ }
+
+ return input.entrySet().stream()
Review Comment:
nit: looks like you just want to remove null value from the map, in which case `Maps.filterValues` could be cleaner.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -316,19 +321,21 @@ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences
@Override
protected Map<KafkaTopicPartition, Long> getRecordLagPerPartition(Map<KafkaTopicPartition, Long> currentOffsets)
{
- if (latestSequenceFromStream == null || currentOffsets == null) {
+ Map<KafkaTopicPartition, Long> endOffsets = offsetSnapshotRef.get().getEndOffsets();
+
+ if (endOffsets == null || currentOffsets == null) {
Review Comment:
same here, can this be null?
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -294,13 +295,17 @@ protected Map<KafkaTopicPartition, Long> getPartitionTimeLag()
// suppress use of CollectionUtils.mapValues() since the valueMapper function is dependent on map key here
@SuppressWarnings("SSBasedInspection")
// Used while calculating cummulative lag for entire stream
- private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(Map<KafkaTopicPartition, Long> currentOffsets)
+ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences()
{
- if (latestSequenceFromStream == null) {
+ OffsetSnapshot<KafkaTopicPartition, Long> offsetSnapshot = offsetSnapshotRef.get();
+ Map<KafkaTopicPartition, Long> currentOffsets = offsetSnapshot.getCurrentOffsets();
+ Map<KafkaTopicPartition, Long> endOffsets = offsetSnapshot.getEndOffsets();
+
+ if (endOffsets == null) {
Review Comment:
Can `offsetSnapshot.getEndOffsets()` be null? I thought you have already replaced null with empty map?
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -172,15 +172,15 @@ protected SeekableStreamSupervisorReportPayload<KafkaTopicPartition, Long> creat
)
{
KafkaSupervisorIOConfig ioConfig = spec.getIoConfig();
- Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences(getHighestCurrentOffsets());
+ Map<KafkaTopicPartition, Long> partitionLag = getRecordLagPerPartitionInLatestSequences();
return new KafkaSupervisorReportPayload(
spec.getId(),
spec.getDataSchema().getDataSource(),
ioConfig.getStream(),
numPartitions,
ioConfig.getReplicas(),
ioConfig.getTaskDuration().getMillis() / 1000,
- includeOffsets ? latestSequenceFromStream : null,
+ includeOffsets ? offsetSnapshotRef.get().getEndOffsets() : null,
Review Comment:
We're replacing `latestSequenceFromStream` with `offsetSnapshotRef` right? Maybe remove `latestSequenceFromStream` as class level variable?
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java:
##########
Review Comment:
Have you considered adding some more complicated test cases to `KafkaSupervisorTest`? I'm not sure if it's feasible, thinking maybe we could use recordSupplier to mock the end_offset, indexerMetadataStorageCoordinator to mock current_offset, and test the lags the system emits. I'm not very familiar with this part of code base, so could be wrong.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -526,6 +539,12 @@ protected void updatePartitionLagFromStream()
latestSequenceFromStream =
partitions.stream().collect(Collectors.toMap(StreamPartition::getPartitionId, recordSupplier::getPosition));
+
+ OffsetSnapshot<KafkaTopicPartition, Long> snapshot = new OffsetSnapshot<>(
Review Comment:
IIUC, we might want to get the current offsets before trying to get the end offset, to minimize the chance of current_offset > end_offset, i.e. move line 544 to 521. I would not call this an atomic update since they're two separate processes, but may be worth call out this in the javadoc for `OffsetSnapshot`, this ensures end_offset > current_offset, and and reduce the issue of negative lag...
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
+ this.endOffsets = toImmutableOffsetMap(endOffsets);
+ }
+
+ private ImmutableMap<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null || input.isEmpty()) {
+ return ImmutableMap.of();
+ }
+
+ return input.entrySet().stream()
+ .filter(e -> e.getValue() != null)
+ .collect(ImmutableMap.toImmutableMap(
+ Map.Entry::getKey,
+ Map.Entry::getValue
+ ));
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getCurrentOffsets()
+ {
+ return currentOffsets;
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getEndOffsets()
+ {
+ return endOffsets;
+ }
+ }
+
+ protected final AtomicReference<OffsetSnapshot<PartitionIdType, SequenceOffsetType>> offsetSnapshotRef =
Review Comment:
Is there any specific reason why you decided to put `offsetSnapshotRef` in `SeekableStreamSupervisor` instead of `KafkaSupervisor`? Maybe it's more suitable there for the sake of encapsulation.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
Review Comment:
Worth adding some javadoc explaining the decision of putting current_offset and end_offset in one class, how it might affect the lag metrics etc...
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -261,14 +261,15 @@ protected List<SeekableStreamIndexTask<KafkaTopicPartition, Long, KafkaRecordEnt
@Override
protected Map<KafkaTopicPartition, Long> getPartitionRecordLag()
{
- Map<KafkaTopicPartition, Long> highestCurrentOffsets = getHighestCurrentOffsets();
+ OffsetSnapshot<KafkaTopicPartition, Long> offsetSnapshot = offsetSnapshotRef.get();
+ Map<KafkaTopicPartition, Long> endOffsets = offsetSnapshot.getEndOffsets();
- if (latestSequenceFromStream == null) {
+ if (endOffsets == null) {
Review Comment:
same here, this be null?
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -294,13 +295,17 @@ protected Map<KafkaTopicPartition, Long> getPartitionTimeLag()
// suppress use of CollectionUtils.mapValues() since the valueMapper function is dependent on map key here
@SuppressWarnings("SSBasedInspection")
// Used while calculating cummulative lag for entire stream
- private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(Map<KafkaTopicPartition, Long> currentOffsets)
+ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences()
{
- if (latestSequenceFromStream == null) {
+ OffsetSnapshot<KafkaTopicPartition, Long> offsetSnapshot = offsetSnapshotRef.get();
+ Map<KafkaTopicPartition, Long> currentOffsets = offsetSnapshot.getCurrentOffsets();
+ Map<KafkaTopicPartition, Long> endOffsets = offsetSnapshot.getEndOffsets();
+
+ if (endOffsets == null) {
return Collections.emptyMap();
}
- return latestSequenceFromStream
+ return endOffsets
Review Comment:
Can `e.getValue()` be null? I thought you have removed all null values from the map. Also, we could use `currentOffsets.getOrDefault` instead of the nullable stuff.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3615174636
@kfaraz @cecemei
Both of your suggestions for improving the code are excellent, and I’m genuinely happy to keep refining it.
Give me a bit of time to rework the design.
Thank you both!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3560869549
@FrankChen021
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3556135697
There was an error when generating the UT report. It looks like a Docker initialization issue. Could someone help take a look? https://github.com/apache/druid/actions/runs/19526187605/job/55903679650
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-4072927121
@FrankChen021 All the issues identified by AI have been fixed—could you take a look?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2621668506
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -535,10 +551,19 @@ protected void updatePartitionLagFromStream()
}
}
+ private void updateOffsetSnapshot(
+ Map<KafkaTopicPartition, Long> highestIngestedOffsets,
+ Map<KafkaTopicPartition, Long> latestOffsetsFromStream
+ )
+ {
+ offsetSnapshotRef.set(
Review Comment:
nit: this could be one line.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3631534267
1. Extract `OffsetSnapshot` into a separate class.
2. Remove the class-level variable `latestSequenceFromStream` from `KafkaSupervisor`.
3. Consolidate all updates and queries to the `offsetSnapshotRef` variable into dedicated methods; other internal functions will no longer access the variable directly.
4. Add Javadoc documentation for the `OffsetSnapshot` class.
5. Incorporate other suggestions from the team.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3578124562
@gianm @capistrant @kfaraz
Can you help review my code and merge it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3680488738
@kfaraz @clintropolis Take a look, please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2756867455
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshotTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetSnapshotTest
+{
+ @Test
+ public void testOffsetSnapshot_emptyInputReturnsEmptyMap()
+ {
+ OffsetSnapshot<String, Long> snapshot = OffsetSnapshot.of(
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ );
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullInputsReturnEmptyMaps()
+ {
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, null);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullCurrentOffsetsReturnsEmptyCurrentMap()
+ {
+ Map<Integer, Long> endOffsets = ImmutableMap.of(0, 100L, 1, 200L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, endOffsets);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(endOffsets, snapshot.getLatestOffsetsFromStream());
Review Comment:
Same issue here: `assertSame(ImmutableMap.of(), ...)` depends on reference identity. Prefer `assertEquals(ImmutableMap.of(), ...)` (and/or `isEmpty`) instead.
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshotTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetSnapshotTest
+{
+ @Test
+ public void testOffsetSnapshot_emptyInputReturnsEmptyMap()
+ {
+ OffsetSnapshot<String, Long> snapshot = OffsetSnapshot.of(
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ );
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullInputsReturnEmptyMaps()
+ {
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, null);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullCurrentOffsetsReturnsEmptyCurrentMap()
+ {
+ Map<Integer, Long> endOffsets = ImmutableMap.of(0, 100L, 1, 200L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, endOffsets);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(endOffsets, snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullEndOffsetsReturnsEmptyEndMap()
+ {
+ Map<Integer, Long> currentOffsets = ImmutableMap.of(0, 50L, 1, 150L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(currentOffsets, null);
+
+ Assert.assertEquals(currentOffsets, snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
Review Comment:
Same issue here: using `assertSame` against `ImmutableMap.of()` is a brittle identity check; use `assertEquals`/`isEmpty` instead.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -506,6 +518,8 @@ protected void updatePartitionLagFromStream()
return;
}
+ Map<KafkaTopicPartition, Long> highestCurrentOffsets = getHighestCurrentOffsets();
+
Review Comment:
The Javadoc immediately above `updatePartitionLagFromStream()` still mentions updating `latestSequenceFromStream`, but that field was removed in favor of the atomic `OffsetSnapshot`/`offsetSnapshotRef`. Please update the comment to avoid pointing readers to a nonexistent symbol and to reflect the new snapshot-based flow.
##########
extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java:
##########
@@ -294,44 +300,48 @@ protected Map<KafkaTopicPartition, Long> getPartitionTimeLag()
// suppress use of CollectionUtils.mapValues() since the valueMapper function is dependent on map key here
@SuppressWarnings("SSBasedInspection")
// Used while calculating cummulative lag for entire stream
- private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(Map<KafkaTopicPartition, Long> currentOffsets)
+ private Map<KafkaTopicPartition, Long> getRecordLagPerPartitionInLatestSequences(OffsetSnapshot<KafkaTopicPartition, Long> offsetSnapshot)
{
- if (latestSequenceFromStream == null) {
- return Collections.emptyMap();
- }
-
- return latestSequenceFromStream
- .entrySet()
- .stream()
- .collect(
- Collectors.toMap(
- Entry::getKey,
- e -> e.getValue() != null
- ? e.getValue() - Optional.ofNullable(currentOffsets.get(e.getKey())).orElse(0L)
- : 0
- )
- );
+ return offsetSnapshot.getLatestOffsetsFromStream()
+ .entrySet()
+ .stream()
+ .collect(Collectors.toMap(
+ Entry::getKey,
+ e -> e.getValue() - offsetSnapshot.getHighestIngestedOffsets().getOrDefault(e.getKey(), 0L)
+ ));
}
+ /**
+ * Computes the record lag for the specified partitions.
+ * <p>
+ * This method is called by the parent class to calculate lag only for the partitions present in
+ * {@code currentOffsets}. The values in {@code currentOffsets} are ignored; only the keys (partitions)
+ * are used to determine which partitions to compute lag for.
+ * <p>
+ * Lag is calculated as {@code latestOffsetFromStream - highestIngestedOffset} using the current atomic
+ * {@link OffsetSnapshot} to ensure consistency between the two offset maps.
+ *
+ * @param currentOffsets map whose keys indicate the partitions to compute lag for (values are ignored)
+ * @return map of partition to its record lag, or empty map if no valid data is available
+ */
@Override
protected Map<KafkaTopicPartition, Long> getRecordLagPerPartition(Map<KafkaTopicPartition, Long> currentOffsets)
{
- if (latestSequenceFromStream == null || currentOffsets == null) {
+ OffsetSnapshot<KafkaTopicPartition, Long> offsetSnapshot = offsetSnapshotRef.get();
+ Map<KafkaTopicPartition, Long> latestSequencesFromStream = offsetSnapshot.getLatestOffsetsFromStream();
+ Map<KafkaTopicPartition, Long> highestIngestedOffsets = offsetSnapshot.getHighestIngestedOffsets();
+
+ if (latestSequencesFromStream.isEmpty() || currentOffsets == null) {
return Collections.emptyMap();
}
- return currentOffsets
- .entrySet()
- .stream()
- .filter(e -> latestSequenceFromStream.get(e.getKey()) != null)
- .collect(
- Collectors.toMap(
- Entry::getKey,
- e -> e.getValue() != null
- ? latestSequenceFromStream.get(e.getKey()) - e.getValue()
- : 0
- )
- );
+ return currentOffsets.keySet().stream()
+ .filter(latestSequencesFromStream::containsKey)
+ .collect(Collectors.toMap(
+ Function.identity(),
+ // compute the lag using offsets from the snapshot.
+ p -> latestSequencesFromStream.get(p) - highestIngestedOffsets.getOrDefault(p, 0L)
+ ));
Review Comment:
`getRecordLagPerPartition` is called by `SeekableStreamSupervisor.generateReport` to compute per-task lag (based on each task's `currentOffsets` values). This implementation ignores those values and instead subtracts the snapshot's *highest* ingested offsets, which can make every task report the same lag and hide lagging replicas. Compute lag using the provided `currentOffsets` values (still using the snapshot only for the latest stream offsets) and update the Javadoc accordingly (including null-value handling to preserve previous behavior).
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshotTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetSnapshotTest
+{
+ @Test
+ public void testOffsetSnapshot_emptyInputReturnsEmptyMap()
+ {
+ OffsetSnapshot<String, Long> snapshot = OffsetSnapshot.of(
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ );
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
Review Comment:
These assertions rely on reference equality with `ImmutableMap.of()`. Even if it happens to be true today, it’s not part of the contract of `ImmutableMap.copyOf(...)` and makes the test brittle across Guava versions/implementations. Prefer `assertEquals(ImmutableMap.of(), ...)` and/or `assertTrue(...isEmpty())`.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshot.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+
+import javax.annotation.Nullable;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Immutable snapshot containing a consistent pair of offset maps: the highest ingested offsets
+ * reported by tasks and the latest end offsets fetched from the underlying stream.
+ *
+ * <p>
+ * The supervisor fetches task-reported ingested offsets first, then fetches end offsets from the stream.
+ * Because these two values are captured at slightly different instants, the reported lag
+ * (latestOffsetsFromStream - highestIngestedOffsets) may be slightly larger than the actual lag at any
+ * precise moment.
+ *
+ * <p>
+ * By publishing both maps together as a single atomic snapshot (using {@link java.util.concurrent.atomic.AtomicReference}),
+ * readers (such as lag metrics and supervisor status) always observe a coherent and consistent view.
+ * This produces stable and monotonic lag trends, avoiding artifacts like temporary negative lags.
Review Comment:
The class Javadoc claims this “produces stable and monotonic lag trends”, but lag is not monotonic in general (it can increase when new records arrive and decrease when ingestion catches up). Consider rewording to focus on “consistent/non-negative snapshots” and avoiding transient negative artifacts, without implying monotonic behavior.
```suggestion
* This provides consistent, non-negative lag snapshots and avoids transient artifacts such as temporary negative lags,
* without implying that lag values themselves are monotonic over time.
```
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshotTest.java:
##########
@@ -0,0 +1,136 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetSnapshotTest
+{
+ @Test
+ public void testOffsetSnapshot_emptyInputReturnsEmptyMap()
+ {
+ OffsetSnapshot<String, Long> snapshot = OffsetSnapshot.of(
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ );
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullInputsReturnEmptyMaps()
+ {
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, null);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
Review Comment:
Same as above: relying on `assertSame(ImmutableMap.of(), ...)` makes the test brittle since reference identity for empty `ImmutableMap` instances isn’t guaranteed by the API contract. Prefer `assertEquals`/`isEmpty` checks here as well.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3596340240
@clintropolis @gianm Take a look, please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3605878850
@wuguowei1994 , thanks for your patience!
We will try to get this PR reviewed soon.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3979834116
@wuguowei1994 let me know once you complete the fix.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3938903240
@FrankChen021 This is chinese new year, give me some time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3584844807
@clintropolis @gianm All checks have passed. Could you please help merge it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3547672163
@amaechler
Could you help me take a look when you have time, please?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2538723924
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final Map<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final Map<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
+ this.endOffsets = toImmutableOffsetMap(endOffsets);
+ }
+
+ private Map<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null || input.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ return input.entrySet().stream()
+ .filter(e -> e.getValue() != null)
+ .collect(ImmutableMap.toImmutableMap(
+ Map.Entry::getKey,
+ Map.Entry::getValue
+ ));
+ }
+
+ public Map<PartitionIdType, SequenceOffsetType> getCurrentOffsets()
+ {
+ return currentOffsets;
+ }
+
+ public Map<PartitionIdType, SequenceOffsetType> getEndOffsets()
Review Comment:
## Exposing internal representation
getEndOffsets exposes the internal representation stored in field endOffsets. The value may be modified [after this call to getEndOffsets](1).
[Show more details](https://github.com/apache/druid/security/code-scanning/10557)
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final Map<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final Map<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
+ this.endOffsets = toImmutableOffsetMap(endOffsets);
+ }
+
+ private Map<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null || input.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ return input.entrySet().stream()
+ .filter(e -> e.getValue() != null)
+ .collect(ImmutableMap.toImmutableMap(
+ Map.Entry::getKey,
+ Map.Entry::getValue
+ ));
+ }
+
+ public Map<PartitionIdType, SequenceOffsetType> getCurrentOffsets()
Review Comment:
## Exposing internal representation
getCurrentOffsets exposes the internal representation stored in field currentOffsets. The value may be modified [after this call to getCurrentOffsets](1).
[Show more details](https://github.com/apache/druid/security/code-scanning/10558)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3558632937
**Can we expedite the process of merging this bug fix?**
Our company is about to upgrade our Druid cluster, and we’d like to include the latest changes from `master` as soon as possible. If we miss this window, the next upgrade won’t happen until next year.
Thank you all for your hard work.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2621685690
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshot.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+
+import javax.annotation.Nullable;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Immutable snapshot containing a consistent pair of offset maps: the highest ingested offsets
+ * reported by tasks and the latest end offsets fetched from the underlying stream.
+ *
+ * <p>
+ * The supervisor fetches task-reported ingested offsets first, then fetches end offsets from the stream.
+ * Because these two values are captured at slightly different instants, the reported lag
+ * (latestOffsetsFromStream - highestIngestedOffsets) may be slightly larger than the actual lag at any
+ * precise moment.
+ *
+ * <p>
+ * By publishing both maps together as a single atomic snapshot (using {@link java.util.concurrent.atomic.AtomicReference}),
+ * readers (such as lag metrics and supervisor status) always observe a coherent and consistent view.
+ * This produces stable and monotonic lag trends, avoiding artifacts like temporary negative lags.
+ *
+ * <p>
+ * This class is generic and can be reused by all seekable-stream supervisors (Kafka, Kinesis, etc.).
+ */
+public final class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+{
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> highestIngestedOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> latestOffsetsFromStream;
+
+ private OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> highestIngestedOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> latestOffsetsFromStream
+ )
+ {
+ this.highestIngestedOffsets = toImmutableOffsetMap(highestIngestedOffsets);
+ this.latestOffsetsFromStream = toImmutableOffsetMap(latestOffsetsFromStream);
+ }
+
+ public static <PartitionIdType, SequenceOffsetType> OffsetSnapshot<PartitionIdType, SequenceOffsetType> of(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ return new OffsetSnapshot<>(currentOffsets, endOffsets);
+ }
+
+ private ImmutableMap<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null) {
+ return ImmutableMap.of();
+ }
+
+ return ImmutableMap.copyOf(Maps.filterValues(input, Objects::nonNull));
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getHighestIngestedOffsets()
Review Comment:
## Exposing internal representation
getHighestIngestedOffsets exposes the internal representation stored in field highestIngestedOffsets. The value may be modified [after this call to getHighestIngestedOffsets](1).
[Show more details](https://github.com/apache/druid/security/code-scanning/10599)
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshotTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetSnapshotTest
+{
+ @Test
+ public void testOffsetSnapshot_emptyInputReturnsEmptyMap()
+ {
+ OffsetSnapshot<String, Long> snapshot = OffsetSnapshot.of(
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ );
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullInputsReturnEmptyMaps()
+ {
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, null);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullCurrentOffsetsReturnsEmptyCurrentMap()
+ {
+ Map<Integer, Long> endOffsets = ImmutableMap.of(0, 100L, 1, 200L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, endOffsets);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(endOffsets, snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullEndOffsetsReturnsEmptyEndMap()
+ {
+ Map<Integer, Long> currentOffsets = ImmutableMap.of(0, 50L, 1, 150L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(currentOffsets, null);
+
+ Assert.assertEquals(currentOffsets, snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_copiesInputMapsAndReturnsImmutableCopies()
+ {
+ Map<String, String> current = new HashMap<>();
+ current.put("p0", "100");
+ current.put("p1", "200");
+
+ Map<String, String> end = new HashMap<>();
+ end.put("p0", "150");
+ end.put("p2", "300");
+
+ OffsetSnapshot<String, String> snapshot = OffsetSnapshot.of(current, end);
+
+ Assert.assertEquals(ImmutableMap.of("p0", "100", "p1", "200"), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(ImmutableMap.of("p0", "150", "p2", "300"), snapshot.getLatestOffsetsFromStream());
+
+ // Returned maps must be immutable
+ Assert.assertThrows(
+ UnsupportedOperationException.class,
+ () -> snapshot.getHighestIngestedOffsets().put("x", "x")
+ );
+ Assert.assertThrows(
+ UnsupportedOperationException.class,
+ () -> snapshot.getLatestOffsetsFromStream().put("x", "x")
Review Comment:
## Deprecated method or constructor invocation
Invoking [ImmutableMap.put](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/10601)
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshot.java:
##########
@@ -0,0 +1,89 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Maps;
+
+import javax.annotation.Nullable;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Immutable snapshot containing a consistent pair of offset maps: the highest ingested offsets
+ * reported by tasks and the latest end offsets fetched from the underlying stream.
+ *
+ * <p>
+ * The supervisor fetches task-reported ingested offsets first, then fetches end offsets from the stream.
+ * Because these two values are captured at slightly different instants, the reported lag
+ * (latestOffsetsFromStream - highestIngestedOffsets) may be slightly larger than the actual lag at any
+ * precise moment.
+ *
+ * <p>
+ * By publishing both maps together as a single atomic snapshot (using {@link java.util.concurrent.atomic.AtomicReference}),
+ * readers (such as lag metrics and supervisor status) always observe a coherent and consistent view.
+ * This produces stable and monotonic lag trends, avoiding artifacts like temporary negative lags.
+ *
+ * <p>
+ * This class is generic and can be reused by all seekable-stream supervisors (Kafka, Kinesis, etc.).
+ */
+public final class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+{
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> highestIngestedOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> latestOffsetsFromStream;
+
+ private OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> highestIngestedOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> latestOffsetsFromStream
+ )
+ {
+ this.highestIngestedOffsets = toImmutableOffsetMap(highestIngestedOffsets);
+ this.latestOffsetsFromStream = toImmutableOffsetMap(latestOffsetsFromStream);
+ }
+
+ public static <PartitionIdType, SequenceOffsetType> OffsetSnapshot<PartitionIdType, SequenceOffsetType> of(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ return new OffsetSnapshot<>(currentOffsets, endOffsets);
+ }
+
+ private ImmutableMap<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null) {
+ return ImmutableMap.of();
+ }
+
+ return ImmutableMap.copyOf(Maps.filterValues(input, Objects::nonNull));
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getHighestIngestedOffsets()
+ {
+ return highestIngestedOffsets;
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getLatestOffsetsFromStream()
Review Comment:
## Exposing internal representation
getLatestOffsetsFromStream exposes the internal representation stored in field latestOffsetsFromStream. The value may be modified [after this call to getLatestOffsetsFromStream](1).
[Show more details](https://github.com/apache/druid/security/code-scanning/10598)
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/OffsetSnapshotTest.java:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream.supervisor;
+
+import com.google.common.collect.ImmutableMap;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+public class OffsetSnapshotTest
+{
+ @Test
+ public void testOffsetSnapshot_emptyInputReturnsEmptyMap()
+ {
+ OffsetSnapshot<String, Long> snapshot = OffsetSnapshot.of(
+ Collections.emptyMap(),
+ Collections.emptyMap()
+ );
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullInputsReturnEmptyMaps()
+ {
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, null);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullCurrentOffsetsReturnsEmptyCurrentMap()
+ {
+ Map<Integer, Long> endOffsets = ImmutableMap.of(0, 100L, 1, 200L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(null, endOffsets);
+
+ Assert.assertTrue(snapshot.getHighestIngestedOffsets().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(endOffsets, snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_nullEndOffsetsReturnsEmptyEndMap()
+ {
+ Map<Integer, Long> currentOffsets = ImmutableMap.of(0, 50L, 1, 150L);
+
+ OffsetSnapshot<Integer, Long> snapshot = OffsetSnapshot.of(currentOffsets, null);
+
+ Assert.assertEquals(currentOffsets, snapshot.getHighestIngestedOffsets());
+ Assert.assertTrue(snapshot.getLatestOffsetsFromStream().isEmpty());
+ Assert.assertSame(ImmutableMap.of(), snapshot.getLatestOffsetsFromStream());
+ }
+
+ @Test
+ public void testOffsetSnapshot_copiesInputMapsAndReturnsImmutableCopies()
+ {
+ Map<String, String> current = new HashMap<>();
+ current.put("p0", "100");
+ current.put("p1", "200");
+
+ Map<String, String> end = new HashMap<>();
+ end.put("p0", "150");
+ end.put("p2", "300");
+
+ OffsetSnapshot<String, String> snapshot = OffsetSnapshot.of(current, end);
+
+ Assert.assertEquals(ImmutableMap.of("p0", "100", "p1", "200"), snapshot.getHighestIngestedOffsets());
+ Assert.assertEquals(ImmutableMap.of("p0", "150", "p2", "300"), snapshot.getLatestOffsetsFromStream());
+
+ // Returned maps must be immutable
+ Assert.assertThrows(
+ UnsupportedOperationException.class,
+ () -> snapshot.getHighestIngestedOffsets().put("x", "x")
Review Comment:
## Deprecated method or constructor invocation
Invoking [ImmutableMap.put](1) should be avoided because it has been deprecated.
[Show more details](https://github.com/apache/druid/security/code-scanning/10600)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3770587927
@gianm Could you help merge the code? We really need this bug fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2588761310
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@ public abstract class SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final ImmutableMap<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
+ this.endOffsets = toImmutableOffsetMap(endOffsets);
+ }
+
+ private ImmutableMap<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null || input.isEmpty()) {
+ return ImmutableMap.of();
+ }
+
+ return input.entrySet().stream()
+ .filter(e -> e.getValue() != null)
+ .collect(ImmutableMap.toImmutableMap(
+ Map.Entry::getKey,
+ Map.Entry::getValue
+ ));
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getCurrentOffsets()
+ {
+ return currentOffsets;
+ }
+
+ public ImmutableMap<PartitionIdType, SequenceOffsetType> getEndOffsets()
+ {
+ return endOffsets;
+ }
+ }
+
+ protected final AtomicReference<OffsetSnapshot<PartitionIdType, SequenceOffsetType>> offsetSnapshotRef =
Review Comment:
@cecemei
`SeekableStreamSupervisor` is the parent class, and it has three subclasses: `KafkaSupervisor`, `KinesisSupervisor`, and `RabbitStreamSupervisor`.
The negative-lag issue mentioned in my PR doesn’t only exist in Kafka — it should also be present in the other two subclasses.
When writing the code, I had the same concern. If we put `offsetSnapshotRef` directly into the `KafkaSupervisor` subclass, then in the future, if someone wants to fix the same issue in `KinesisSupervisor` or `RabbitStreamSupervisor`, they would need to redefine `offsetSnapshotRef` in each subclass.
**The ideal approach would be:**
1. Define `offsetSnapshotRef` in the parent class.
2. Fix the negative-lag issue in all three subclasses.
If needed, I can update the changes in this way. What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "cecemei (via GitHub)" <gi...@apache.org>.
cecemei commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3613883194
> Thanks for the changes, @wuguowei1994 ! I have left some suggestions.
>
> While the changes in this PR make sense by reporting the lag more consistently (updating the two offsets in lockstep), I was wondering if it wouldn't be simpler to just report zero lag in case the lag turns out to be negative.
>
> A negative record lag does mean that the task has already caught up to the last offsets that we had fetched from the topic and ingested some more records beyond that. And the lag metric is really just meant to indicate if the tasks are keeping up.
>
> For other purposes, we have the message gap and the time lag metrics.
>
> In fact, the negative lag could even be a feature to identify if some tasks are particularly slow in returning their offsets. 😛 , and we could probably have alerts set up if the negative lag goes below a specific threshold.
>
> @cecemei , I think you also raised a concern regarding the possibility that the lag reported might now be higher than the actual lag, since we always fetch the stream offsets only after we have received updates from all the tasks.
>
> I think the current code is also susceptible to reporting stale lag (higher or lower).
>
> Say if a task were slow to return its latest ingested offsets, we would be delayed in fetching the latest offsets from the stream. So, in that period, we would be reporting stale lag (which could have been higher or lower than the actual lag, a special case of which would be negative lag) and then as soon as we fetched the latest offsets from the stream, the reported lag would fix itself.
Yes I suspect we might seeing a slightly higher lag after this change (comparing with what might be reported before). The accuracy of the lag would not be affected by when the `latestSequenceFromStream` gets updated (more random), but would only be affected by how long it took fetching the offsets (delay inside the system and interaction with kafka). I do think it provides more consistency than before and the trend of the lag is more important, so it's an improvement.
For future reference, we could maybe calculate the lag on a per partition basis.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on code in PR #18750:
URL: https://github.com/apache/druid/pull/18750#discussion_r2540224025
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final Map<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final Map<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
+ this.endOffsets = toImmutableOffsetMap(endOffsets);
+ }
+
+ private Map<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null || input.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ return input.entrySet().stream()
+ .filter(e -> e.getValue() != null)
+ .collect(ImmutableMap.toImmutableMap(
+ Map.Entry::getKey,
+ Map.Entry::getValue
+ ));
+ }
+
+ public Map<PartitionIdType, SequenceOffsetType> getCurrentOffsets()
Review Comment:
fixed
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -177,6 +178,50 @@
// Internal data structures
// --------------------------------------------------------
+ protected static class OffsetSnapshot<PartitionIdType, SequenceOffsetType>
+ {
+ private final Map<PartitionIdType, SequenceOffsetType> currentOffsets;
+ private final Map<PartitionIdType, SequenceOffsetType> endOffsets;
+
+ public OffsetSnapshot(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> currentOffsets,
+ @Nullable Map<PartitionIdType, SequenceOffsetType> endOffsets
+ )
+ {
+ this.currentOffsets = toImmutableOffsetMap(currentOffsets);
+ this.endOffsets = toImmutableOffsetMap(endOffsets);
+ }
+
+ private Map<PartitionIdType, SequenceOffsetType> toImmutableOffsetMap(
+ @Nullable Map<PartitionIdType, SequenceOffsetType> input
+ )
+ {
+ if (input == null || input.isEmpty()) {
+ return Collections.emptyMap();
+ }
+
+ return input.entrySet().stream()
+ .filter(e -> e.getValue() != null)
+ .collect(ImmutableMap.toImmutableMap(
+ Map.Entry::getKey,
+ Map.Entry::getValue
+ ));
+ }
+
+ public Map<PartitionIdType, SequenceOffsetType> getCurrentOffsets()
+ {
+ return currentOffsets;
+ }
+
+ public Map<PartitionIdType, SequenceOffsetType> getEndOffsets()
Review Comment:
fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3555815548
@abhishekrb19 Thanks for your patient response. I've synced the latest master into my branch and rerun the UTs locally. Everything looks good now.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3714551605
Take a look? Jan 6th
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "FrankChen021 (via GitHub)" <gi...@apache.org>.
FrankChen021 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3844897074
@wuguowei1994 Could you resolve the comments from copilot?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Fix negative Kafka partition lag caused by inconsistent current/latest offsets (druid)
Posted by "wuguowei1994 (via GitHub)" <gi...@apache.org>.
wuguowei1994 commented on PR #18750:
URL: https://github.com/apache/druid/pull/18750#issuecomment-3706814065
Take a look ? Jan 3th
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org