This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch 38.0.0
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/38.0.0 by this push:
new 3fc1f6dddc4 fix: Fix invalid time comparison in
SqlSegmentsMetadataManager (#19603)
3fc1f6dddc4 is described below
commit 3fc1f6dddc4ba30839d2caf4fa393c29e9443809
Author: Vivek Dhiman <[email protected]>
AuthorDate: Sat Aug 1 20:21:56 2026 +0530
fix: Fix invalid time comparison in SqlSegmentsMetadataManager (#19603)
Changes:
- Use timestamp in nanos to compare poll finish time and method invocation
time
- Avoid duplicate invocations of OnDemandDatabasePoll while waiting
(cherry picked from commit 5542a09d68167db57ce8022f9a3498d6790c369a)
---
.../druid/metadata/SqlSegmentsMetadataManager.java | 46 +++++++++++++++-------
1 file changed, 32 insertions(+), 14 deletions(-)
diff --git
a/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataManager.java
b/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataManager.java
index d18ad5a0b2a..57ccc9d8cf3 100644
---
a/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataManager.java
+++
b/server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataManager.java
@@ -90,7 +90,7 @@ public class SqlSegmentsMetadataManager implements
SegmentsMetadataManager
* leadership changes.
*/
final CompletableFuture<Void> firstPollCompletionFuture = new
CompletableFuture<>();
- long lastPollStartTimestampInMs = -1;
+ Long lastPollStartTimestampInNanos = null;
}
/**
@@ -101,12 +101,25 @@ public class SqlSegmentsMetadataManager implements
SegmentsMetadataManager
static class OnDemandDatabasePoll implements DatabasePoll
{
final long initiationTimeNanos = System.nanoTime();
- final CompletableFuture<Void> pollCompletionFuture = new
CompletableFuture<>();
+ final CompletableFuture<Long> pollCompletionFuture = new
CompletableFuture<>();
long nanosElapsedFromInitiation()
{
return System.nanoTime() - initiationTimeNanos;
}
+
+ boolean hasCompletedSuccessfullyAfter(long timestampNanons)
+ {
+ return pollCompletionFuture.isDone()
+ && !pollCompletionFuture.isCancelled()
+ && !pollCompletionFuture.isCompletedExceptionally()
+ && Futures.getUnchecked(pollCompletionFuture) > timestampNanons;
+ }
+
+ void waitToFinish()
+ {
+ Futures.getUnchecked(pollCompletionFuture);
+ }
}
/**
@@ -429,7 +442,7 @@ public class SqlSegmentsMetadataManager implements
SegmentsMetadataManager
lock.lock();
try {
if (startOrder == currentStartPollingOrder) {
- periodicDatabasePoll.lastPollStartTimestampInMs =
System.currentTimeMillis();
+ periodicDatabasePoll.lastPollStartTimestampInNanos =
System.nanoTime();
poll();
periodicDatabasePoll.firstPollCompletionFuture.complete(null);
latestDatabasePoll = periodicDatabasePoll;
@@ -549,29 +562,34 @@ public class SqlSegmentsMetadataManager implements
SegmentsMetadataManager
/**
* This method will always force a database poll if there is no ongoing
database poll. This method will then
- * waits for the new poll or the ongoing poll to completes before returning.
+ * wait for the new poll or the ongoing poll to complete before returning.
* This means that any method using this check can be sure that the latest
poll for the snapshot was completed after
* this method was called.
*/
@VisibleForTesting
void forceOrWaitOngoingDatabasePoll()
{
- long checkStartTime = System.currentTimeMillis();
+ long checkStartTimeNanos = System.nanoTime();
ReentrantReadWriteLock.WriteLock lock = startStopPollLock.writeLock();
lock.lock();
try {
DatabasePoll latestDatabasePoll = this.latestDatabasePoll;
try {
//Verify if there was a periodic poll completed while we were waiting
for the lock
- if (latestDatabasePoll instanceof PeriodicDatabasePoll
- && ((PeriodicDatabasePoll)
latestDatabasePoll).lastPollStartTimestampInMs > checkStartTime) {
- return;
+ if (latestDatabasePoll instanceof PeriodicDatabasePoll
latestPeriodicPoll) {
+ if (latestPeriodicPoll.lastPollStartTimestampInNanos != null
+ && latestPeriodicPoll.lastPollStartTimestampInNanos >
checkStartTimeNanos) {
+ return;
+ }
}
- // Verify if there was a on-demand poll completed while we were
waiting for the lock
- if (latestDatabasePoll instanceof OnDemandDatabasePoll) {
- long checkStartTimeNanos =
TimeUnit.MILLISECONDS.toNanos(checkStartTime);
- OnDemandDatabasePoll latestOnDemandPoll = (OnDemandDatabasePoll)
latestDatabasePoll;
- if (latestOnDemandPoll.initiationTimeNanos > checkStartTimeNanos) {
+ // Verify if there was an on-demand poll completed while we were
waiting for the lock
+ if (latestDatabasePoll instanceof OnDemandDatabasePoll
latestOnDemandPoll) {
+ if
(latestOnDemandPoll.hasCompletedSuccessfullyAfter(checkStartTimeNanos)) {
+ // This poll completed while we were waiting for the lock
+ return;
+ } else if (latestOnDemandPoll.initiationTimeNanos >
checkStartTimeNanos) {
+ // This poll started while we were waiting for the lock
+ latestOnDemandPoll.waitToFinish();
return;
}
}
@@ -594,7 +612,7 @@ public class SqlSegmentsMetadataManager implements
SegmentsMetadataManager
{
try {
poll();
- onDemandPoll.pollCompletionFuture.complete(null);
+ onDemandPoll.pollCompletionFuture.complete(System.nanoTime());
}
catch (Throwable t) {
onDemandPoll.pollCompletionFuture.completeExceptionally(t);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
