This is an automated email from the ASF dual-hosted git repository.
kfaraz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new 2c2a1c1d681 fix: Try register upgraded pending segment on all
supervisors for datasource (#19679)
2c2a1c1d681 is described below
commit 2c2a1c1d68180f836121e6cca63e88d0134d2e38
Author: Kashif Faraz <[email protected]>
AuthorDate: Tue Jul 14 13:20:48 2026 +0530
fix: Try register upgraded pending segment on all supervisors for
datasource (#19679)
Changes:
- Return a List of matching supervisor IDs from `getActiveSupervisorId`
- Try to register the ugpraded pending segment on all matching supervisors
- For supervisors which are not aware of this pending segment, the
registration would be a no-op
---
.../actions/SegmentTransactionalReplaceAction.java | 75 +++++++++++-----------
.../overlord/supervisor/SupervisorManager.java | 12 ++--
.../supervisor/SeekableStreamSupervisor.java | 9 ++-
.../SegmentTransactionalReplaceActionTest.java | 21 +++---
.../ConcurrentReplaceAndStreamingAppendTest.java | 5 +-
.../overlord/supervisor/SupervisorManagerTest.java | 18 +++---
6 files changed, 73 insertions(+), 67 deletions(-)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java
index d1dc2909faa..b4ad8accac5 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java
@@ -23,7 +23,6 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.core.type.TypeReference;
import com.google.common.annotations.VisibleForTesting;
-import com.google.common.base.Optional;
import com.google.common.collect.ImmutableSet;
import org.apache.druid.indexing.common.SegmentUpgradeMetrics;
import org.apache.druid.indexing.common.task.IndexTaskUtils;
@@ -183,6 +182,10 @@ public class SegmentTransactionalReplaceAction implements
TaskAction<SegmentPubl
List<PendingSegmentRecord> upgradedPendingSegments
)
{
+ if (upgradedPendingSegments == null || upgradedPendingSegments.isEmpty()) {
+ return;
+ }
+
// Emit one persisted event per upgraded segment (rather than a single
aggregate) regardless of whether a supervisor
// exists to receive them, so the total can be compared against the count
actually announced by tasks and so each
// event carries the segment's interval and version.
@@ -194,10 +197,10 @@ public class SegmentTransactionalReplaceAction implements
TaskAction<SegmentPubl
}
final SupervisorManager supervisorManager = toolbox.getSupervisorManager();
- final Optional<String> activeSupervisorIdWithAppendLock =
-
supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(task.getDataSource());
+ final List<String> activeSupervisorIdWithAppendLock =
+
supervisorManager.getActiveSupervisorIdsForDatasourceWithAppendLock(task.getDataSource());
- if (!activeSupervisorIdWithAppendLock.isPresent()) {
+ if (activeSupervisorIdWithAppendLock.isEmpty()) {
log.info(
"Could not find any active concurrent streaming supervisor for
datasource[%s]."
+ " Ignoring registry of [%d] pending segment(s) upgraded by
task[%s]. These will become queryable only"
@@ -209,49 +212,47 @@ public class SegmentTransactionalReplaceAction implements
TaskAction<SegmentPubl
return;
}
- // Register each upgraded pending segment on the supervisor and summarize
the batch. The per-segment mapping is
- // unbounded (a broad REPLACE can upgrade thousands of segments), so INFO
carries aggregate counts plus a bounded
- // sample and the full mapping is materialized only when DEBUG is enabled.
- final boolean debugEnabled = log.isDebugEnabled();
- final Map<String, Integer> notifiedTasksSample = new LinkedHashMap<>();
- final Map<String, Integer> notifiedTasksBySegment = debugEnabled ? new
LinkedHashMap<>() : null;
- int registeredSegments = 0;
- int totalNotifiedTasks = 0;
+ // Try to register the upgraded pending segments with all eligible
supervisors
+ // Only the supervisor which owns a pending segment will actually register
it
+ for (String supervisorId : activeSupervisorIdWithAppendLock) {
+ registerUpgradedPendingSegmentOnSupervisor(task, toolbox,
upgradedPendingSegments, supervisorId);
+ }
+ }
+
+ /**
+ * Registers each upgraded pending segment on the given supervisor and prints
+ * a summary of the batch.
+ */
+ private void registerUpgradedPendingSegmentOnSupervisor(
+ Task task,
+ TaskActionToolbox toolbox,
+ List<PendingSegmentRecord> upgradedPendingSegments,
+ String supervisorId
+ )
+ {
+ final Map<String, Integer> notifiedTasksBySegment = new LinkedHashMap<>();
for (PendingSegmentRecord upgradedPendingSegment :
upgradedPendingSegments) {
- final OptionalInt notified = supervisorManager
-
.registerUpgradedPendingSegmentOnSupervisor(activeSupervisorIdWithAppendLock.get(),
upgradedPendingSegment);
+ final OptionalInt notified = toolbox.getSupervisorManager()
+ .registerUpgradedPendingSegmentOnSupervisor(supervisorId,
upgradedPendingSegment);
if (notified.isEmpty()) {
continue;
}
final int notifiedCount = notified.getAsInt();
- registeredSegments++;
- totalNotifiedTasks += notifiedCount;
- if (notifiedTasksSample.size() < NOTIFIED_LOG_SAMPLE_SIZE) {
- notifiedTasksSample.put(upgradedPendingSegment.getId().toString(),
notifiedCount);
- }
- if (debugEnabled) {
- notifiedTasksBySegment.put(upgradedPendingSegment.getId().toString(),
notifiedCount);
- }
+ notifiedTasksBySegment.put(upgradedPendingSegment.getId().toString(),
notifiedCount);
}
+ final boolean truncateSummary = notifiedTasksBySegment.size() >
NOTIFIED_LOG_SAMPLE_SIZE && !log.isDebugEnabled();
log.info(
- "Registered [%d] upgraded pending segment(s) created by task[%s] on
supervisor[%s], notifying [%d] running"
- + " task(s) in total. Tasks notified per segment%s: %s",
- registeredSegments,
+ "Registered [%d] upgraded pending segment(s) created by task[%s] on
supervisor[%s]."
+ + " Tasks notified per segment %s: %s",
+ notifiedTasksBySegment.size(),
task.getId(),
- activeSupervisorIdWithAppendLock.get(),
- totalNotifiedTasks,
- registeredSegments > NOTIFIED_LOG_SAMPLE_SIZE ? " (first " +
NOTIFIED_LOG_SAMPLE_SIZE + ", enable debug for all)" : "",
- notifiedTasksSample
+ supervisorId,
+ truncateSummary ? " (first " + NOTIFIED_LOG_SAMPLE_SIZE + ", enable
debug for all)" : "",
+ truncateSummary
+ ?
notifiedTasksBySegment.entrySet().stream().limit(NOTIFIED_LOG_SAMPLE_SIZE)
+ : notifiedTasksBySegment
);
- if (debugEnabled && registeredSegments > NOTIFIED_LOG_SAMPLE_SIZE) {
- log.debug(
- "Tasks notified per segment created by task[%s] on supervisor[%s]:
%s",
- task.getId(),
- activeSupervisorIdWithAppendLock.get(),
- notifiedTasksBySegment
- );
- }
}
@Override
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
index 929f34bf89e..d20fb3cfe79 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
@@ -125,11 +125,15 @@ public class SupervisorManager implements
SupervisorStatsProvider
}
/**
+ * Finds the list of active streaming supervisors ingest into the given
datasource
+ * using an APPEND lock.
+ *
* @param datasource Datasource to find active supervisor id with append
lock for.
- * @return An optional with the active appending supervisor id if it exists.
+ * @return List of active appending supervisor IDs, if any.
*/
- public Optional<String>
getActiveSupervisorIdForDatasourceWithAppendLock(String datasource)
+ public List<String> getActiveSupervisorIdsForDatasourceWithAppendLock(String
datasource)
{
+ final List<String> matchingSupervisorIds = new ArrayList<>();
for (Map.Entry<String, Pair<Supervisor, SupervisorSpec>> entry :
supervisors.entrySet()) {
final String supervisorId = entry.getKey();
final Supervisor supervisor = entry.getValue().lhs;
@@ -142,11 +146,11 @@ public class SupervisorManager implements
SupervisorStatsProvider
&& !supervisorSpec.isSuspended()
&& supervisorSpec.getDataSources().contains(datasource)
&& (hasAppendLock)) {
- return Optional.of(supervisorId);
+ matchingSupervisorIds.add(supervisorId);
}
}
- return Optional.absent();
+ return matchingSupervisorIds;
}
public Optional<SupervisorSpec> getSupervisorSpec(String id)
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
index ecc4ef3d00b..5b6220fb246 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
@@ -1450,9 +1450,12 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
}
if (notifiedTasks == 0) {
- // No running task matched: the segment will not be re-announced until
handoff. This is a potential silent-loss
- // window where data will not be queryable until handoff.
- log.warn(
+ // No running task matched: the segment will not be re-announced until
handoff.
+ // This may happen if there are multiple supervisors for the same
datasource
+ // or due to race conditions while events such as changing the state of
a TaskGroup
+ // from actively reading to pending completion, etc.
+ // This is a potential silent-loss window where data will not be
queryable until handoff.
+ log.info(
"Could not find any task matching taskAllocatorId[%s] in
supervisor[%s] for upgraded pending segment[%s]"
+ " (upgradedFrom[%s]); it will not be re-announced until handoff.",
taskAllocatorId,
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceActionTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceActionTest.java
index 069763d9be2..6a25885fe73 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceActionTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceActionTest.java
@@ -19,7 +19,6 @@
package org.apache.druid.indexing.common.actions;
-import com.google.common.base.Optional;
import com.google.common.collect.ImmutableSet;
import org.apache.druid.indexing.common.SegmentUpgradeMetrics;
import org.apache.druid.indexing.common.task.NoopTask;
@@ -73,8 +72,8 @@ public class SegmentTransactionalReplaceActionTest
@Test
public void testNoActiveSupervisorStillEmitsPersistedPerSegment()
{
-
EasyMock.expect(supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(DATA_SOURCE))
- .andReturn(Optional.absent());
+
EasyMock.expect(supervisorManager.getActiveSupervisorIdsForDatasourceWithAppendLock(DATA_SOURCE))
+ .andReturn(List.of());
// registerUpgradedPendingSegmentOnSupervisor must not be called when
there is no active supervisor.
EasyMock.replay(toolbox, supervisorManager);
@@ -87,8 +86,8 @@ public class SegmentTransactionalReplaceActionTest
@Test
public void testActiveSupervisorRegistersEachSegment()
{
-
EasyMock.expect(supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(DATA_SOURCE))
- .andReturn(Optional.of(SUPERVISOR_ID));
+
EasyMock.expect(supervisorManager.getActiveSupervisorIdsForDatasourceWithAppendLock(DATA_SOURCE))
+ .andReturn(List.of(SUPERVISOR_ID));
EasyMock.expect(
supervisorManager.registerUpgradedPendingSegmentOnSupervisor(EasyMock.eq(SUPERVISOR_ID),
EasyMock.anyObject())
).andReturn(OptionalInt.of(2)).times(3);
@@ -103,8 +102,8 @@ public class SegmentTransactionalReplaceActionTest
@Test
public void testBatchLargerThanSampleSizeRegistersEverySegment()
{
-
EasyMock.expect(supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(DATA_SOURCE))
- .andReturn(Optional.of(SUPERVISOR_ID));
+
EasyMock.expect(supervisorManager.getActiveSupervisorIdsForDatasourceWithAppendLock(DATA_SOURCE))
+ .andReturn(List.of(SUPERVISOR_ID));
EasyMock.expect(
supervisorManager.registerUpgradedPendingSegmentOnSupervisor(EasyMock.eq(SUPERVISOR_ID),
EasyMock.anyObject())
).andReturn(OptionalInt.of(1)).times(MORE_THAN_SAMPLE_SIZE);
@@ -121,8 +120,8 @@ public class SegmentTransactionalReplaceActionTest
{
// An empty result models a non-seekable-stream supervisor or a
registration failure; the segment is still counted
// as persisted but contributes no notified-task summary.
-
EasyMock.expect(supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(DATA_SOURCE))
- .andReturn(Optional.of(SUPERVISOR_ID));
+
EasyMock.expect(supervisorManager.getActiveSupervisorIdsForDatasourceWithAppendLock(DATA_SOURCE))
+ .andReturn(List.of(SUPERVISOR_ID));
EasyMock.expect(
supervisorManager.registerUpgradedPendingSegmentOnSupervisor(EasyMock.eq(SUPERVISOR_ID),
EasyMock.anyObject())
).andReturn(OptionalInt.empty()).times(2);
@@ -140,8 +139,8 @@ public class SegmentTransactionalReplaceActionTest
final Level originalLevel =
LogManager.getLogger(SegmentTransactionalReplaceAction.class).getLevel();
Configurator.setLevel(SegmentTransactionalReplaceAction.class.getName(),
Level.DEBUG);
try {
-
EasyMock.expect(supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(DATA_SOURCE))
- .andReturn(Optional.of(SUPERVISOR_ID));
+
EasyMock.expect(supervisorManager.getActiveSupervisorIdsForDatasourceWithAppendLock(DATA_SOURCE))
+ .andReturn(List.of(SUPERVISOR_ID));
EasyMock.expect(
supervisorManager.registerUpgradedPendingSegmentOnSupervisor(EasyMock.eq(SUPERVISOR_ID),
EasyMock.anyObject())
).andReturn(OptionalInt.of(1)).times(MORE_THAN_SAMPLE_SIZE);
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java
index 35382746ae1..a57f98fee7f 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java
@@ -19,7 +19,6 @@
package org.apache.druid.indexing.common.task.concurrent;
-import com.google.common.base.Optional;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Iterables;
@@ -130,8 +129,8 @@ public class ConcurrentReplaceAndStreamingAppendTest
extends IngestionTestBase
public void setUpIngestionTestBase() throws IOException
{
EasyMock.reset(supervisorManager);
-
EasyMock.expect(supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(TestDataSource.WIKI))
- .andReturn(Optional.of(TestDataSource.WIKI)).anyTimes();
+
EasyMock.expect(supervisorManager.getActiveSupervisorIdsForDatasourceWithAppendLock(TestDataSource.WIKI))
+ .andReturn(List.of(TestDataSource.WIKI)).anyTimes();
super.setUpIngestionTestBase();
final TaskConfig taskConfig = new TaskConfigBuilder().build();
taskActionClientFactory = createActionClientFactory();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
index 7bbedd87fb8..23e9f94fa8b 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
@@ -780,7 +780,7 @@ public class SupervisorManagerTest extends EasyMockSupport
}
@Test
- public void testGetActiveSupervisorIdForDatasourceWithAppendLock()
+ public void testGetActiveSupervisorIdsForDatasourceWithAppendLock()
{
EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(Collections.emptyMap());
@@ -860,26 +860,26 @@ public class SupervisorManagerTest extends EasyMockSupport
replayAll();
manager.start();
-
Assert.assertFalse(manager.getActiveSupervisorIdForDatasourceWithAppendLock("nonExistent").isPresent());
+
Assert.assertTrue(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("nonExistent").isEmpty());
manager.createOrUpdateAndStartSupervisor(noopSupervisorSpec);
-
Assert.assertFalse(manager.getActiveSupervisorIdForDatasourceWithAppendLock("noopDS").isPresent());
+
Assert.assertTrue(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("noopDS").isEmpty());
manager.createOrUpdateAndStartSupervisor(suspendedSpec);
-
Assert.assertFalse(manager.getActiveSupervisorIdForDatasourceWithAppendLock("suspendedDS").isPresent());
+
Assert.assertTrue(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("suspendedDS").isEmpty());
manager.createOrUpdateAndStartSupervisor(activeSpec);
-
Assert.assertFalse(manager.getActiveSupervisorIdForDatasourceWithAppendLock("activeDS").isPresent());
+
Assert.assertTrue(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("activeDS").isEmpty());
manager.createOrUpdateAndStartSupervisor(activeAppendSpec);
-
Assert.assertTrue(manager.getActiveSupervisorIdForDatasourceWithAppendLock("activeAppendDS").isPresent());
+
Assert.assertFalse(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("activeAppendDS").isEmpty());
manager.createOrUpdateAndStartSupervisor(activeSpecWithConcurrentLocks);
-
Assert.assertTrue(manager.getActiveSupervisorIdForDatasourceWithAppendLock("activeConcurrentLocksDS").isPresent());
+
Assert.assertFalse(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("activeConcurrentLocksDS").isEmpty());
manager.createOrUpdateAndStartSupervisor(specWithUseConcurrentLocksFalse);
- Assert.assertFalse(
-
manager.getActiveSupervisorIdForDatasourceWithAppendLock("dsWithUseConcurrentLocksFalse").isPresent()
+ Assert.assertTrue(
+
manager.getActiveSupervisorIdsForDatasourceWithAppendLock("dsWithUseConcurrentLocksFalse").isEmpty()
);
verifyAll();
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
