This is an automated email from the ASF dual-hosted git repository.
capistrant 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 f6b4c708ae9 minor: Add logs and metrics observability to the segment
upgrade path that fires while using concurrent locking (#19651)
f6b4c708ae9 is described below
commit f6b4c708ae9d9d70ed4270e982640e360678d85c
Author: Lucas Capistrant <[email protected]>
AuthorDate: Sun Jul 12 15:10:39 2026 -0500
minor: Add logs and metrics observability to the segment upgrade path that
fires while using concurrent locking (#19651)
* Add logs and metrics observability to the segment upgrade path for
concurrent append and replace ingestion
mets checkpoint
doc and test
tidy up
self review
* Apply suggestions from code review
applying the pure text changes. will commit responses to review in another
commit
Co-authored-by: Kashif Faraz <[email protected]>
* upgrade metrics names
* review responses
* fixup statsd emitter json
* dropwizard json def update
* detailed task action logging
* more review comments
* coverage
---------
Co-authored-by: Kashif Faraz <[email protected]>
---
docs/operations/metrics.md | 6 +
.../main/resources/defaultMetricDimensions.json | 54 ++++++
.../src/main/resources/defaultMetrics.json | 36 ++++
.../src/main/resources/defaultMetrics.json | 6 +
.../main/resources/defaultMetricDimensions.json | 7 +
.../indexing/common/SegmentUpgradeMetrics.java | 64 ++++++++
.../actions/SegmentTransactionalReplaceAction.java | 81 ++++++++-
.../druid/indexing/common/task/IndexTaskUtils.java | 18 ++
.../overlord/supervisor/SupervisorManager.java | 13 +-
.../SeekableStreamIndexTaskRunner.java | 61 ++++++-
.../supervisor/SeekableStreamSupervisor.java | 84 +++++++++-
.../SegmentTransactionalReplaceActionTest.java | 181 +++++++++++++++++++++
.../ConcurrentReplaceAndStreamingAppendTest.java | 3 +-
.../overlord/supervisor/SupervisorManagerTest.java | 8 +-
.../SeekableStreamSupervisorStateTest.java | 80 +++++++++
.../java/org/apache/druid/query/DruidMetrics.java | 1 +
.../realtime/appenderator/StreamAppenderator.java | 49 +++++-
.../appenderator/StreamAppenderatorTest.java | 54 ++++++
18 files changed, 781 insertions(+), 25 deletions(-)
diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md
index f4757760b71..81f8f21b917 100644
--- a/docs/operations/metrics.md
+++ b/docs/operations/metrics.md
@@ -317,6 +317,12 @@ batch ingestion emit the following metrics. These metrics
are deltas for each em
|`ingest/notices/time`|Milliseconds taken to process a notice by the
supervisor.|`supervisorId`, `dataSource`, `tags`| < 1s |
|`ingest/pause/time`|Milliseconds spent by a task in a paused state without
ingesting.|`dataSource`, `taskId`, `tags`| < 10 seconds|
|`ingest/handoff/time`|Total number of milliseconds taken to handoff a set of
segments.|`dataSource`, `taskId`, `taskType`, `groupId`, `tags`|Depends on the
coordinator cycle time.|
+|`ingest/realtime/segmentUpgrade/persisted`|Number of pending segments that
were upgraded in the metadata store while publishing segments to an interval.
Emitted by the Overlord only when a REPLACE task (e.g., compaction task with
`useConcurrentLocks` set to `true`) commits segments to an interval already
containing pending segments allocated to an APPEND task (e.g. streaming task
with `useConcurrentLocks` set to `true`).|`dataSource`, `taskId`, `taskType`,
`groupId`, `interval`, `versio [...]
+|`ingest/realtime/segmentUpgrade/notified`|Number of notifications the
supervisor sent to running tasks, emitted once per task so it scales with the
replica count. Reconcile against the per-task outcomes: `notified` should equal
`announced` + `skipped` + `sendFailed` over the same period and `dataSource`,
since every notified task either announces, skips, or fails to receive the
request. A shortfall indicates a notification was silently
dropped.|`supervisorId`, `dataSource`, `stream`, `t [...]
+|`ingest/realtime/segmentUpgrade/unmatched`|Number of upgraded pending
segments the supervisor could not route to any running task. These are not
announced under the new version until handoff, so the corresponding data may be
briefly missing from queries.|`supervisorId`, `dataSource`, `stream`,
`interval`, `version`, `tags`|0. A non-zero value indicates a lost upgrade
announcement.|
+|`ingest/realtime/segmentUpgrade/sendFailed`|Number of upgrade requests that
matched a running task but failed to reach it over the wire after
retries.|`supervisorId`, `dataSource`, `stream`, `taskId`, `interval`,
`version`, `tags`|0|
+|`ingest/realtime/segmentUpgrade/announced`|Number of upgraded segments a task
announced under the new version. Emitted once per task, so it scales with the
replica count. Reconcile against `notified` (also per task), not
`ingest/realtime/segmentUpgrade/persisted`, which is per segment rather than
per replica.|`dataSource`, `taskId`, `taskType`, `groupId`, `interval`,
`version`, `tags`|Greater than 0 while concurrent replace occurs.|
+|`ingest/realtime/segmentUpgrade/skipped`|Number of upgrade requests a task
received but did not announce. The `reason` dimension is one of `unknown base`
(the request reached the wrong task), `base sink already dropped` (the base
sink is no longer present), or `dropping base sink` (the base sink is handing
off, which is benign and covered by the durable publish path).|`dataSource`,
`taskId`, `taskType`, `groupId`, `interval`, `version`, `tags`, `reason`|0,
excluding `reason=dropping bas [...]
|`task/autoScaler/requiredCount`|Count of required tasks based on the
calculations of the auto scaler.|`supervisorId`, `dataSource`, `stream`,
`scalingSkipReason`|Depends on auto scaler config.|
|`task/autoScaler/scaleActionTime`|Time taken in milliseconds to complete the
scale action.|`supervisorId`, `dataSource`, `stream`, `tags`|Depends on auto
scaler config.|
|`task/autoScaler/costBased/optimalTaskCount`|Optimal task count computed by
the cost-based auto scaler.|`supervisorId`, `dataSource`, `stream`|Depends on
auto scaler config.|
diff --git
a/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json
b/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json
index 3a681d25cd9..0c83023448e 100644
---
a/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json
+++
b/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json
@@ -214,6 +214,60 @@
"type": "timer",
"timeUnit": "NANOSECONDS"
},
+ "ingest/realtime/segmentUpgrade/persisted": {
+ "dimensions": [
+ "dataSource",
+ "interval",
+ "version"
+ ],
+ "type": "counter"
+ },
+ "ingest/realtime/segmentUpgrade/notified": {
+ "dimensions": [
+ "dataSource",
+ "stream",
+ "taskId",
+ "interval",
+ "version"
+ ],
+ "type": "counter"
+ },
+ "ingest/realtime/segmentUpgrade/unmatched": {
+ "dimensions": [
+ "dataSource",
+ "stream",
+ "interval",
+ "version"
+ ],
+ "type": "counter"
+ },
+ "ingest/realtime/segmentUpgrade/sendFailed": {
+ "dimensions": [
+ "dataSource",
+ "stream",
+ "taskId",
+ "interval",
+ "version"
+ ],
+ "type": "counter"
+ },
+ "ingest/realtime/segmentUpgrade/announced": {
+ "dimensions": [
+ "dataSource",
+ "interval",
+ "version"
+ ],
+ "type": "counter"
+ },
+ "ingest/realtime/segmentUpgrade/skipped": {
+ "dimensions": [
+ "dataSource",
+ "reason",
+ "interval",
+ "version"
+ ],
+ "type": "counter"
+ },
"task/run/time": {
"dimensions": [
"dataSource",
diff --git
a/extensions-contrib/opentsdb-emitter/src/main/resources/defaultMetrics.json
b/extensions-contrib/opentsdb-emitter/src/main/resources/defaultMetrics.json
index f38f115348f..0a1b237b5f0 100644
--- a/extensions-contrib/opentsdb-emitter/src/main/resources/defaultMetrics.json
+++ b/extensions-contrib/opentsdb-emitter/src/main/resources/defaultMetrics.json
@@ -105,6 +105,42 @@
"ingest/events/messageGap": [
"dataSource"
],
+ "ingest/realtime/segmentUpgrade/persisted": [
+ "dataSource",
+ "interval",
+ "version"
+ ],
+ "ingest/realtime/segmentUpgrade/notified": [
+ "dataSource",
+ "stream",
+ "taskId",
+ "interval",
+ "version"
+ ],
+ "ingest/realtime/segmentUpgrade/unmatched": [
+ "dataSource",
+ "stream",
+ "interval",
+ "version"
+ ],
+ "ingest/realtime/segmentUpgrade/sendFailed": [
+ "dataSource",
+ "stream",
+ "taskId",
+ "interval",
+ "version"
+ ],
+ "ingest/realtime/segmentUpgrade/announced": [
+ "dataSource",
+ "interval",
+ "version"
+ ],
+ "ingest/realtime/segmentUpgrade/skipped": [
+ "dataSource",
+ "reason",
+ "interval",
+ "version"
+ ],
"ingest/kafka/lag": [
"dataSource",
"stream"
diff --git
a/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
b/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
index 52d72cf19bc..cd7b609bb58 100644
---
a/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
+++
b/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json
@@ -134,6 +134,12 @@
"ingest/notices/time" : { "dimensions" : ["dataSource"], "type" : "timer",
"conversionFactor": 1000.0, "help": "Seconds taken to process a notice by the
supervisor." },
"ingest/pause/time" : { "dimensions" : ["dataSource"], "type" : "timer",
"conversionFactor": 1000.0, "help": "Seconds spent by a task in a paused state
without ingesting." },
"ingest/handoff/time" : { "dimensions" : ["dataSource"], "type" : "timer",
"conversionFactor": 1000.0, "help": "Total number of seconds taken to handoff a
set of segments." },
+ "ingest/realtime/segmentUpgrade/persisted" : { "dimensions" :
["dataSource"], "type" : "count", "help": "Number of pending segments upgraded
by a concurrent replace." },
+ "ingest/realtime/segmentUpgrade/notified" : { "dimensions" : ["dataSource",
"stream"], "type" : "count", "help": "Number of notifications the supervisor
sent to running tasks, one per task." },
+ "ingest/realtime/segmentUpgrade/unmatched" : { "dimensions" : ["dataSource",
"stream"], "type" : "count", "help": "Number of upgraded pending segments that
matched no running task." },
+ "ingest/realtime/segmentUpgrade/sendFailed" : { "dimensions" :
["dataSource", "stream"], "type" : "count", "help": "Number of upgrade requests
that failed to reach a task." },
+ "ingest/realtime/segmentUpgrade/announced" : { "dimensions" :
["dataSource"], "type" : "count", "help": "Number of upgraded segments a task
announced under the new version." },
+ "ingest/realtime/segmentUpgrade/skipped" : { "dimensions" : ["dataSource",
"reason"], "type" : "count", "help": "Number of upgrade requests a task
received but did not announce." },
"task/autoScaler/requiredCount" : { "dimensions" : ["dataSource"], "type" :
"count", "help": "Count of required tasks based on the calculations of lagBased
auto scaler." },
"task/run/time" : { "dimensions" : ["dataSource", "taskType"], "type" :
"timer", "conversionFactor": 1000.0, "help": "Seconds taken to run a task."},
diff --git
a/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json
b/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json
index 24a0129362e..226ec036ec7 100644
---
a/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json
+++
b/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json
@@ -57,6 +57,13 @@
"ingest/segments/count" : { "dimensions" : ["dataSource"], "type" : "count"
},
"ingest/rows/published": { "dimensions" : ["dataSource"], "type" : "count" },
+ "ingest/realtime/segmentUpgrade/persisted" : { "dimensions" : ["dataSource",
"interval", "version"], "type" : "count" },
+ "ingest/realtime/segmentUpgrade/notified" : { "dimensions" : ["dataSource",
"stream", "taskId", "interval", "version"], "type" : "count" },
+ "ingest/realtime/segmentUpgrade/unmatched" : { "dimensions" : ["dataSource",
"stream", "interval", "version"], "type" : "count" },
+ "ingest/realtime/segmentUpgrade/sendFailed" : { "dimensions" :
["dataSource", "stream", "taskId", "interval", "version"], "type" : "count" },
+ "ingest/realtime/segmentUpgrade/announced" : { "dimensions" : ["dataSource",
"interval", "version"], "type" : "count" },
+ "ingest/realtime/segmentUpgrade/skipped" : { "dimensions" : ["dataSource",
"reason", "interval", "version"], "type" : "count" },
+
"ingest/kafka/lag" : { "dimensions" : ["dataSource", "stream"], "type" :
"gauge" },
"ingest/kafka/maxLag" : { "dimensions" : ["dataSource", "stream"], "type" :
"gauge" },
"ingest/kafka/avgLag" : { "dimensions" : ["dataSource", "stream"], "type" :
"gauge" },
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentUpgradeMetrics.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentUpgradeMetrics.java
new file mode 100644
index 00000000000..1b64fb74740
--- /dev/null
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentUpgradeMetrics.java
@@ -0,0 +1,64 @@
+/*
+ * 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.common;
+
+/**
+ * Metric names and dimension values for the re-announcement of pending
segments upgraded by a concurrent REPLACE.
+ * <ul>
+ * <li>the task action emits {@link #PERSISTED} (how many upgrades a commit
persisted),</li>
+ * <li>the supervisor emits {@link #NOTIFIED}, {@link #UNMATCHED} and {@link
#SEND_FAILED} as it fans requests out,</li>
+ * <li>the streaming task emits {@link #ANNOUNCED} and {@link #SKIPPED}
(with a {@code reason}) as it applies them.</li>
+ * </ul>
+ * Two reconciliations bound the visibility gap:
+ * <ul>
+ * <li>per segment, {@link #UNMATCHED} counts upgrades that reached no
running task (delayed until handoff);</li>
+ * <li>per task, {@link #NOTIFIED} should equal {@link #ANNOUNCED} + {@link
#SKIPPED} + {@link #SEND_FAILED},
+ * since every notified task either announces, skips, or fails to
receive the request. A shortfall means a
+ * notification was silently dropped.</li>
+ * </ul>
+ */
+public class SegmentUpgradeMetrics
+{
+ /** Number of upgraded pending segments a REPLACE commit persisted and
handed to the supervisor. Task-action dims. */
+ public static final String PERSISTED =
"ingest/realtime/segmentUpgrade/persisted";
+
+ /** A notification was sent to a running task (once per task). Supervisor
dims plus {@code taskId}. */
+ public static final String NOTIFIED =
"ingest/realtime/segmentUpgrade/notified";
+
+ /** A record matched no running task and will not be re-announced until
handoff. Supervisor dims. */
+ public static final String UNMATCHED =
"ingest/realtime/segmentUpgrade/unmatched";
+
+ /** An upgrade request failed to reach a task over the wire. Supervisor dims
plus {@code taskId}. */
+ public static final String SEND_FAILED =
"ingest/realtime/segmentUpgrade/sendFailed";
+
+ /** A task announced an upgraded segment under the new version. Task dims. */
+ public static final String ANNOUNCED =
"ingest/realtime/segmentUpgrade/announced";
+
+ /**
+ * A task received an upgrade request but did not announce it. The {@code
reason} dimension carries
+ * {@link
org.apache.druid.segment.realtime.appenderator.StreamAppenderator.PendingSegmentUpgradeResult#getReason()}.
+ * Task dims.
+ */
+ public static final String SKIPPED =
"ingest/realtime/segmentUpgrade/skipped";
+
+ private SegmentUpgradeMetrics()
+ {
+ }
+}
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 6546ed80d9d..d1dc2909faa 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
@@ -22,14 +22,17 @@ package org.apache.druid.indexing.common.actions;
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;
import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.indexing.overlord.CriticalAction;
import org.apache.druid.indexing.overlord.SegmentPublishResult;
import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
import org.apache.druid.metadata.PendingSegmentRecord;
import org.apache.druid.metadata.ReplaceTaskLock;
import org.apache.druid.segment.SegmentSchemaMapping;
@@ -37,7 +40,10 @@ import org.apache.druid.segment.SegmentUtils;
import org.apache.druid.timeline.DataSegment;
import javax.annotation.Nullable;
+import java.util.LinkedHashMap;
import java.util.List;
+import java.util.Map;
+import java.util.OptionalInt;
import java.util.Set;
import java.util.stream.Collectors;
@@ -63,6 +69,13 @@ public class SegmentTransactionalReplaceAction implements
TaskAction<SegmentPubl
{
private static final Logger log = new
Logger(SegmentTransactionalReplaceAction.class);
+ /**
+ * Upper bound on the number of per-segment entries included in the INFO
summary of upgraded pending segments. A broad
+ * REPLACE (e.g. compaction over a wide interval of sparse,
finely-partitioned append data) can upgrade thousands of
+ * pending segments, so the full mapping is reserved for DEBUG.
+ */
+ private static final int NOTIFIED_LOG_SAMPLE_SIZE = 10;
+
/**
* Set of segments to be inserted into metadata storage
*/
@@ -163,26 +176,82 @@ public class SegmentTransactionalReplaceAction implements
TaskAction<SegmentPubl
/**
* Registers upgraded pending segments on the active supervisor, if any
*/
- private void registerUpgradedPendingSegmentsOnSupervisor(
+ @VisibleForTesting
+ void registerUpgradedPendingSegmentsOnSupervisor(
Task task,
TaskActionToolbox toolbox,
List<PendingSegmentRecord> upgradedPendingSegments
)
{
+ // 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.
+ for (PendingSegmentRecord upgradedPendingSegment :
upgradedPendingSegments) {
+ final ServiceMetricEvent.Builder metricBuilder = new
ServiceMetricEvent.Builder();
+ IndexTaskUtils.setTaskDimensions(metricBuilder, task);
+ IndexTaskUtils.setPendingSegmentDimensions(metricBuilder,
upgradedPendingSegment);
+
toolbox.getEmitter().emit(metricBuilder.setMetric(SegmentUpgradeMetrics.PERSISTED,
1));
+ }
+
final SupervisorManager supervisorManager = toolbox.getSupervisorManager();
final Optional<String> activeSupervisorIdWithAppendLock =
supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(task.getDataSource());
if (!activeSupervisorIdWithAppendLock.isPresent()) {
+ 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"
+ + " after handoff.",
+ task.getDataSource(),
+ upgradedPendingSegments.size(),
+ task.getId()
+ );
return;
}
- upgradedPendingSegments.forEach(
- upgradedPendingSegment ->
supervisorManager.registerUpgradedPendingSegmentOnSupervisor(
- activeSupervisorIdWithAppendLock.get(),
- upgradedPendingSegment
- )
+ // 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;
+ for (PendingSegmentRecord upgradedPendingSegment :
upgradedPendingSegments) {
+ final OptionalInt notified = supervisorManager
+
.registerUpgradedPendingSegmentOnSupervisor(activeSupervisorIdWithAppendLock.get(),
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);
+ }
+ }
+
+ 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,
+ task.getId(),
+ activeSupervisorIdWithAppendLock.get(),
+ totalNotifiedTasks,
+ registeredSegments > NOTIFIED_LOG_SAMPLE_SIZE ? " (first " +
NOTIFIED_LOG_SAMPLE_SIZE + ", enable debug for all)" : "",
+ notifiedTasksSample
);
+ 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/common/task/IndexTaskUtils.java
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
index 29692e1aa98..8f719c2cc39 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/common/task/IndexTaskUtils.java
@@ -25,7 +25,9 @@ import
org.apache.druid.indexing.overlord.SegmentPublishResult;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.emitter.service.SegmentMetadataEvent;
import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.PendingSegmentRecord;
import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
import org.apache.druid.timeline.DataSegment;
import java.util.Collection;
@@ -119,4 +121,20 @@ public class IndexTaskUtils
.mapToLong(Integer::longValue)
.sum();
}
+
+ /**
+ * Adds the upgraded pending segment's {@code interval} and {@code version}
to a metric builder so that every
+ * segment-upgrade metric can be sliced by the specific segment being
re-announced. Mirrors
+ * {@code IndexTaskUtils.setSegmentDimensions}, which serves the same
purpose for {@code DataSegment}s.
+ */
+ public static ServiceMetricEvent.Builder setPendingSegmentDimensions(
+ ServiceMetricEvent.Builder metricBuilder,
+ PendingSegmentRecord pendingSegmentRecord
+ )
+ {
+ final SegmentIdWithShardSpec id = pendingSegmentRecord.getId();
+ return metricBuilder
+ .setDimension(DruidMetrics.INTERVAL, id.getInterval().toString())
+ .setDimension(DruidMetrics.VERSION, id.getVersion());
+ }
}
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 fa7d96634ae..929f34bf89e 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
@@ -60,6 +60,7 @@ import java.util.Collection;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.OptionalInt;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;
@@ -511,8 +512,11 @@ public class SupervisorManager implements
SupervisorStatsProvider
* Registers a new version of the given pending segment on a supervisor. This
* allows the supervisor to include the pending segment in queries fired
against
* that segment version.
+ *
+ * @return the number of tasks notified if the segment was registered on a
seekable stream supervisor, or
+ * {@link OptionalInt#empty()} if no such supervisor was found or the
registration failed
*/
- public boolean registerUpgradedPendingSegmentOnSupervisor(
+ public OptionalInt registerUpgradedPendingSegmentOnSupervisor(
String supervisorId,
PendingSegmentRecord upgradedPendingSegment
)
@@ -529,12 +533,11 @@ public class SupervisorManager implements
SupervisorStatsProvider
Pair<Supervisor, SupervisorSpec> supervisor =
supervisors.get(supervisorId);
Preconditions.checkNotNull(supervisor, "supervisor could not be found");
if (!(supervisor.lhs instanceof SeekableStreamSupervisor)) {
- return false;
+ return OptionalInt.empty();
}
SeekableStreamSupervisor<?, ?, ?> seekableStreamSupervisor =
(SeekableStreamSupervisor<?, ?, ?>) supervisor.lhs;
-
seekableStreamSupervisor.registerNewVersionOfPendingSegment(upgradedPendingSegment);
- return true;
+ return
OptionalInt.of(seekableStreamSupervisor.registerNewVersionOfPendingSegment(upgradedPendingSegment));
}
catch (Exception e) {
log.error(
@@ -545,7 +548,7 @@ public class SupervisorManager implements
SupervisorStatsProvider
supervisorId
);
}
- return false;
+ return OptionalInt.empty();
}
/**
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
index 2390792fd4d..d2054550fd6 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java
@@ -58,6 +58,7 @@ import
org.apache.druid.indexer.report.IngestionStatsAndErrorsTaskReport;
import org.apache.druid.indexer.report.TaskContextReport;
import org.apache.druid.indexer.report.TaskReport;
import org.apache.druid.indexing.common.LockGranularity;
+import org.apache.druid.indexing.common.SegmentUpgradeMetrics;
import org.apache.druid.indexing.common.TaskLock;
import org.apache.druid.indexing.common.TaskLockType;
import org.apache.druid.indexing.common.TaskToolbox;
@@ -80,6 +81,7 @@ import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.concurrent.Execs;
import org.apache.druid.java.util.emitter.EmittingLogger;
import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.query.DruidMetrics;
import org.apache.druid.segment.incremental.InputRowFilterResult;
import org.apache.druid.segment.incremental.ParseExceptionHandler;
import org.apache.druid.segment.incremental.ParseExceptionReport;
@@ -1846,7 +1848,9 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
{
authorizationCheck(req);
try {
- ((StreamAppenderator)
appenderator).registerUpgradedPendingSegment(upgradedPendingSegment);
+ final StreamAppenderator.PendingSegmentUpgradeResult outcome =
+ ((StreamAppenderator)
appenderator).registerUpgradedPendingSegment(upgradedPendingSegment);
+ recordUpgradeResult(upgradedPendingSegment, outcome);
return Response.ok().build();
}
catch (DruidException e) {
@@ -1865,6 +1869,61 @@ public abstract class
SeekableStreamIndexTaskRunner<PartitionIdType, SequenceOff
}
}
+ /**
+ * Logs and emits a metric for the result of a pending-segment upgrade
request, keyed off the {@code result}
+ * returned by {@link StreamAppenderator#registerUpgradedPendingSegment}.
+ */
+ private void recordUpgradeResult(
+ PendingSegmentRecord upgradedPendingSegment,
+ StreamAppenderator.PendingSegmentUpgradeResult result
+ )
+ {
+ final SegmentIdWithShardSpec upgradedId = upgradedPendingSegment.getId();
+ final String upgradedFromSegmentId =
upgradedPendingSegment.getUpgradedFromSegmentId();
+
+ if (result == StreamAppenderator.PendingSegmentUpgradeResult.ANNOUNCED) {
+ toolbox.getEmitter().emit(
+ IndexTaskUtils.setPendingSegmentDimensions(task.getMetricBuilder(),
upgradedPendingSegment)
+ .setMetric(SegmentUpgradeMetrics.ANNOUNCED, 1)
+ );
+ return;
+ }
+ // Log at the level appropriate to the result; the reason string itself is
carried by the enum.
+ switch (result) {
+ case SKIPPED_UNKNOWN_BASE:
+ // The request targeted the wrong task: this task never held a base
sink matching upgradedFromSegmentId.
+ log.warn(
+ "Not announcing upgraded pending segment[%s] on task[%s] because
it has no base sink matching"
+ + " upgradedFromSegmentId[%s]; the upgrade request likely targeted
the wrong task.",
+ upgradedId, task.getId(), upgradedFromSegmentId
+ );
+ break;
+ case SKIPPED_NO_SINK:
+ // Unexpected: the base sink is gone even though this task once held
it.
+ log.info(
+ "Not announcing upgraded pending segment[%s] (upgradedFrom[%s]) on
task[%s] because the base sink is no"
+ + " longer present.",
+ upgradedId, upgradedFromSegmentId, task.getId()
+ );
+ break;
+ case SKIPPED_DROPPING:
+ // Expected during handoff: the base sink is being dropped and the
durable path re-announces at the new version.
+ log.debug(
+ "Not announcing upgraded pending segment[%s] (upgradedFrom[%s]) on
task[%s] because the base sink is being"
+ + " dropped.",
+ upgradedId, upgradedFromSegmentId, task.getId()
+ );
+ break;
+ default:
+ return;
+ }
+ toolbox.getEmitter().emit(
+ IndexTaskUtils.setPendingSegmentDimensions(task.getMetricBuilder(),
upgradedPendingSegment)
+ .setDimension(DruidMetrics.REASON, result.getReason())
+ .setMetric(SegmentUpgradeMetrics.SKIPPED, 1)
+ );
+ }
+
public Map<String, Object> doGetRowStats()
{
Map<String, Object> returnMap = new HashMap<>();
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 74329c68e1d..ecc4ef3d00b 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
@@ -32,6 +32,7 @@ import com.google.common.collect.ImmutableMap;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Sets;
import com.google.common.util.concurrent.AsyncFunction;
+import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.ListeningExecutorService;
@@ -48,7 +49,9 @@ import org.apache.druid.error.InvalidInput;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskState;
import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexing.common.SegmentUpgradeMetrics;
import org.apache.druid.indexing.common.TaskInfoProvider;
+import org.apache.druid.indexing.common.task.IndexTaskUtils;
import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.indexing.overlord.DataSourceMetadata;
import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
@@ -1415,26 +1418,97 @@ public abstract class
SeekableStreamSupervisor<PartitionIdType, SequenceOffsetTy
addNotice(new ResetOffsetsNotice(resetDataSourceMetadata));
}
- public void registerNewVersionOfPendingSegment(
+ /**
+ * Notifies every running task in the matching task group(s) of an upgraded
pending segment.
+ *
+ * @return the number of tasks notified
+ */
+ public int registerNewVersionOfPendingSegment(
PendingSegmentRecord pendingSegmentRecord
)
{
+ final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId();
+ int notifiedTasks = 0;
+
for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) {
- if
(taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) {
+ if (taskGroup.baseSequenceName.equals(taskAllocatorId)) {
for (String taskId : taskGroup.taskIds()) {
- taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord);
+ notifyTaskOfUpgradedPendingSegment(taskId, pendingSegmentRecord);
+ notifiedTasks++;
}
}
}
for (List<TaskGroup> taskGroupList : pendingCompletionTaskGroups.values())
{
for (TaskGroup taskGroup : taskGroupList) {
- if
(taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) {
+ if (taskGroup.baseSequenceName.equals(taskAllocatorId)) {
for (String taskId : taskGroup.taskIds()) {
- taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord);
+ notifyTaskOfUpgradedPendingSegment(taskId, pendingSegmentRecord);
+ notifiedTasks++;
}
}
}
}
+
+ 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(
+ "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,
+ supervisorId,
+ pendingSegmentRecord.getId(),
+ pendingSegmentRecord.getUpgradedFromSegmentId()
+ );
+ emitter.emit(
+ IndexTaskUtils.setPendingSegmentDimensions(getMetricBuilder(),
pendingSegmentRecord)
+ .setMetric(SegmentUpgradeMetrics.UNMATCHED, 1)
+ );
+ }
+ return notifiedTasks;
+ }
+
+ /**
+ * Sends an upgraded pending segment to a single task, emitting {@link
SegmentUpgradeMetrics#NOTIFIED} for the
+ * notification attempt and {@link SegmentUpgradeMetrics#SEND_FAILED} if the
request fails to reach the task. Both
+ * are per-task (keyed by {@code taskId}), so {@code notified} can be
reconciled against the per-task
+ * {@code announced}, {@code skipped} and {@code sendFailed} outcomes.
+ */
+ private void notifyTaskOfUpgradedPendingSegment(String taskId,
PendingSegmentRecord pendingSegmentRecord)
+ {
+ emitter.emit(
+ IndexTaskUtils.setPendingSegmentDimensions(getMetricBuilder(),
pendingSegmentRecord)
+ .setDimension(DruidMetrics.TASK_ID, taskId)
+ .setMetric(SegmentUpgradeMetrics.NOTIFIED, 1)
+ );
+ Futures.addCallback(
+ taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord),
+ new FutureCallback<>()
+ {
+ @Override
+ public void onSuccess(Boolean result)
+ {
+ // The request reached the task; what the task does with it
(announce or skip) is reported by the task's
+ // own announced/skipped metrics, so there is nothing to record
here.
+ }
+
+ @Override
+ public void onFailure(Throwable t)
+ {
+ log.warn(
+ t,
+ "Failed to register upgraded pending segment[%s] on task[%s]
of supervisor[%s].",
+ pendingSegmentRecord.getId(), taskId, supervisorId
+ );
+ emitter.emit(
+ IndexTaskUtils.setPendingSegmentDimensions(getMetricBuilder(),
pendingSegmentRecord)
+ .setDimension(DruidMetrics.TASK_ID, taskId)
+ .setMetric(SegmentUpgradeMetrics.SEND_FAILED, 1)
+ );
+ }
+ },
+ MoreExecutors.directExecutor()
+ );
}
public ReentrantLock getRecordSupplierLock()
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
new file mode 100644
index 00000000000..069763d9be2
--- /dev/null
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceActionTest.java
@@ -0,0 +1,181 @@
+/*
+ * 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.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;
+import org.apache.druid.indexing.common.task.Task;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.metrics.StubServiceEmitter;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
+import org.apache.druid.timeline.partition.NumberedShardSpec;
+import org.apache.logging.log4j.Level;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.config.Configurator;
+import org.easymock.EasyMock;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.OptionalInt;
+
+public class SegmentTransactionalReplaceActionTest
+{
+ private static final String DATA_SOURCE = "wiki";
+ private static final String SUPERVISOR_ID = "sup1";
+
+ // Larger than SegmentTransactionalReplaceAction.NOTIFIED_LOG_SAMPLE_SIZE
(10) so the bounded/truncated log path runs.
+ private static final int MORE_THAN_SAMPLE_SIZE = 12;
+
+ private final SegmentTransactionalReplaceAction action =
+ SegmentTransactionalReplaceAction.create(ImmutableSet.of(), null);
+
+ private StubServiceEmitter emitter;
+ private SupervisorManager supervisorManager;
+ private TaskActionToolbox toolbox;
+ private Task task;
+
+ @Before
+ public void setUp()
+ {
+ emitter = new StubServiceEmitter("test", "localhost");
+ supervisorManager = EasyMock.createMock(SupervisorManager.class);
+ toolbox = EasyMock.createMock(TaskActionToolbox.class);
+ task = NoopTask.forDatasource(DATA_SOURCE);
+
+ EasyMock.expect(toolbox.getEmitter()).andReturn(emitter).anyTimes();
+
EasyMock.expect(toolbox.getSupervisorManager()).andReturn(supervisorManager).anyTimes();
+ }
+
+ @Test
+ public void testNoActiveSupervisorStillEmitsPersistedPerSegment()
+ {
+
EasyMock.expect(supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(DATA_SOURCE))
+ .andReturn(Optional.absent());
+ // registerUpgradedPendingSegmentOnSupervisor must not be called when
there is no active supervisor.
+ EasyMock.replay(toolbox, supervisorManager);
+
+ action.registerUpgradedPendingSegmentsOnSupervisor(task, toolbox,
records(2));
+
+ Assert.assertEquals(2,
emitter.getMetricEventCount(SegmentUpgradeMetrics.PERSISTED));
+ EasyMock.verify(supervisorManager);
+ }
+
+ @Test
+ public void testActiveSupervisorRegistersEachSegment()
+ {
+
EasyMock.expect(supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(DATA_SOURCE))
+ .andReturn(Optional.of(SUPERVISOR_ID));
+ EasyMock.expect(
+
supervisorManager.registerUpgradedPendingSegmentOnSupervisor(EasyMock.eq(SUPERVISOR_ID),
EasyMock.anyObject())
+ ).andReturn(OptionalInt.of(2)).times(3);
+ EasyMock.replay(toolbox, supervisorManager);
+
+ action.registerUpgradedPendingSegmentsOnSupervisor(task, toolbox,
records(3));
+
+ Assert.assertEquals(3,
emitter.getMetricEventCount(SegmentUpgradeMetrics.PERSISTED));
+ EasyMock.verify(supervisorManager);
+ }
+
+ @Test
+ public void testBatchLargerThanSampleSizeRegistersEverySegment()
+ {
+
EasyMock.expect(supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(DATA_SOURCE))
+ .andReturn(Optional.of(SUPERVISOR_ID));
+ EasyMock.expect(
+
supervisorManager.registerUpgradedPendingSegmentOnSupervisor(EasyMock.eq(SUPERVISOR_ID),
EasyMock.anyObject())
+ ).andReturn(OptionalInt.of(1)).times(MORE_THAN_SAMPLE_SIZE);
+ EasyMock.replay(toolbox, supervisorManager);
+
+ action.registerUpgradedPendingSegmentsOnSupervisor(task, toolbox,
records(MORE_THAN_SAMPLE_SIZE));
+
+ Assert.assertEquals(MORE_THAN_SAMPLE_SIZE,
emitter.getMetricEventCount(SegmentUpgradeMetrics.PERSISTED));
+ EasyMock.verify(supervisorManager);
+ }
+
+ @Test
+ public void testSegmentsNotRegisteredOnSupervisorAreStillPersisted()
+ {
+ // 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.registerUpgradedPendingSegmentOnSupervisor(EasyMock.eq(SUPERVISOR_ID),
EasyMock.anyObject())
+ ).andReturn(OptionalInt.empty()).times(2);
+ EasyMock.replay(toolbox, supervisorManager);
+
+ action.registerUpgradedPendingSegmentsOnSupervisor(task, toolbox,
records(2));
+
+ Assert.assertEquals(2,
emitter.getMetricEventCount(SegmentUpgradeMetrics.PERSISTED));
+ EasyMock.verify(supervisorManager);
+ }
+
+ @Test
+ public void testFullPerSegmentDetailLoggedAtDebug()
+ {
+ 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.registerUpgradedPendingSegmentOnSupervisor(EasyMock.eq(SUPERVISOR_ID),
EasyMock.anyObject())
+ ).andReturn(OptionalInt.of(1)).times(MORE_THAN_SAMPLE_SIZE);
+ EasyMock.replay(toolbox, supervisorManager);
+
+ action.registerUpgradedPendingSegmentsOnSupervisor(task, toolbox,
records(MORE_THAN_SAMPLE_SIZE));
+
+ Assert.assertEquals(MORE_THAN_SAMPLE_SIZE,
emitter.getMetricEventCount(SegmentUpgradeMetrics.PERSISTED));
+ EasyMock.verify(supervisorManager);
+ }
+ finally {
+ Configurator.setLevel(SegmentTransactionalReplaceAction.class.getName(),
originalLevel);
+ }
+ }
+
+ private static List<PendingSegmentRecord> records(int count)
+ {
+ final List<PendingSegmentRecord> records = new ArrayList<>(count);
+ for (int i = 0; i < count; i++) {
+ records.add(
+ PendingSegmentRecord.create(
+ new SegmentIdWithShardSpec(
+ DATA_SOURCE,
+ Intervals.of("2024/2025"),
+ "v1",
+ new NumberedShardSpec(i, 0)
+ ),
+ "sequenceName",
+ "prevId" + i,
+ "upgradedFrom" + i,
+ "taskAllocatorId"
+ )
+ );
+ }
+ return records;
+ }
+}
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 868d8c15568..35382746ae1 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
@@ -84,6 +84,7 @@ import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.OptionalInt;
import java.util.Set;
import java.util.concurrent.atomic.AtomicInteger;
@@ -170,7 +171,7 @@ public class ConcurrentReplaceAndStreamingAppendTest
extends IngestionTestBase
EasyMock.expect(supervisorManager.registerUpgradedPendingSegmentOnSupervisor(
EasyMock.capture(supervisorId),
EasyMock.capture(pendingSegment)
- )).andReturn(true).anyTimes();
+ )).andReturn(OptionalInt.of(1)).anyTimes();
replaceTask = createAndStartTask();
EasyMock.replay(supervisorManager);
versionToIntervalToLoadSpecs = new HashMap<>();
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 36e87b604d3..7bbedd87fb8 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
@@ -75,6 +75,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.OptionalInt;
import java.util.concurrent.ConcurrentHashMap;
@RunWith(EasyMockRunner.class)
@@ -894,8 +895,7 @@ public class SupervisorManagerTest extends EasyMockSupport
SeekableStreamSupervisorSpec streamingSpec =
EasyMock.createNiceMock(SeekableStreamSupervisorSpec.class);
SeekableStreamSupervisor streamSupervisor =
EasyMock.createNiceMock(SeekableStreamSupervisor.class);
- streamSupervisor.registerNewVersionOfPendingSegment(EasyMock.anyObject());
- EasyMock.expectLastCall().once();
+
EasyMock.expect(streamSupervisor.registerNewVersionOfPendingSegment(EasyMock.anyObject())).andReturn(1).once();
EasyMock.expect(streamingSpec.getId()).andReturn("sss").anyTimes();
EasyMock.expect(streamingSpec.isSuspended()).andReturn(false).anyTimes();
EasyMock.expect(streamingSpec.getDataSources()).andReturn(ImmutableList.of("DS")).anyTimes();
@@ -923,10 +923,10 @@ public class SupervisorManagerTest extends EasyMockSupport
manager.start();
manager.createOrUpdateAndStartSupervisor(noopSpec);
-
Assert.assertFalse(manager.registerUpgradedPendingSegmentOnSupervisor("noop",
pendingSegment));
+
Assert.assertFalse(manager.registerUpgradedPendingSegmentOnSupervisor("noop",
pendingSegment).isPresent());
manager.createOrUpdateAndStartSupervisor(streamingSpec);
-
Assert.assertTrue(manager.registerUpgradedPendingSegmentOnSupervisor("sss",
pendingSegment));
+ Assert.assertEquals(OptionalInt.of(1),
manager.registerUpgradedPendingSegmentOnSupervisor("sss", pendingSegment));
verifyAll();
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
index 9e45920ad71..6c12c8b3885 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
@@ -39,6 +39,7 @@ import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.indexer.TaskLocation;
import org.apache.druid.indexer.TaskStatus;
import org.apache.druid.indexer.granularity.UniformGranularitySpec;
+import org.apache.druid.indexing.common.SegmentUpgradeMetrics;
import org.apache.druid.indexing.common.TestUtils;
import org.apache.druid.indexing.common.task.Task;
import org.apache.druid.indexing.overlord.DataSourceMetadata;
@@ -2116,6 +2117,85 @@ public class SeekableStreamSupervisorStateTest extends
EasyMockSupport
Assert.assertEquals(pendingSegmentRecord0, captured0.getValue());
Assert.assertEquals(pendingSegmentRecord1, captured1.getValue());
+
+ // Both records matched a running task, so each emits a notified metric
and none is unmatched.
+ Assert.assertEquals(2,
emitter.getMetricEventCount(SegmentUpgradeMetrics.NOTIFIED));
+ Assert.assertEquals(0,
emitter.getMetricEventCount(SegmentUpgradeMetrics.UNMATCHED));
+ verifyAll();
+ }
+
+ @Test
+ public void testRegisterNewVersionOfPendingSegmentUnmatchedEmitsMetric()
+ {
+ EasyMock.expect(spec.isSuspended()).andReturn(false);
+ replayAll();
+
+ final TestSeekableStreamSupervisor supervisor = new
TestSeekableStreamSupervisor();
+ supervisor.getIoConfig().setTaskCount(3);
+ supervisor.start();
+
+ supervisor.addTaskGroupToActivelyReadingTaskGroup(
+ supervisor.getTaskGroupIdForPartition("0"),
+ ImmutableMap.of("0", "5"),
+ null,
+ null,
+ Set.of("task0"),
+ Set.of(),
+ null
+ );
+
+ // A taskAllocatorId that matches no running task group
+ final PendingSegmentRecord record = PendingSegmentRecord.create(
+ new SegmentIdWithShardSpec("DS", Intervals.of("2024/2025"), "2024",
new NumberedShardSpec(1, 0)),
+ "no-such-allocator",
+ "prevId",
+ "someAppendedSegment",
+ "no-such-allocator"
+ );
+
+ supervisor.registerNewVersionOfPendingSegment(record);
+
+ Assert.assertEquals(1,
emitter.getMetricEventCount(SegmentUpgradeMetrics.UNMATCHED));
+ Assert.assertEquals(0,
emitter.getMetricEventCount(SegmentUpgradeMetrics.NOTIFIED));
+ verifyAll();
+ }
+
+ @Test
+ public void testRegisterNewVersionOfPendingSegmentSendFailureEmitsMetric()
+ {
+ EasyMock.expect(spec.isSuspended()).andReturn(false);
+ EasyMock.expect(
+
indexTaskClient.registerNewVersionOfPendingSegmentAsync(EasyMock.eq("task0"),
EasyMock.anyObject())
+ ).andReturn(Futures.immediateFailedFuture(new RuntimeException("boom")));
+ replayAll();
+
+ final TestSeekableStreamSupervisor supervisor = new
TestSeekableStreamSupervisor();
+ supervisor.getIoConfig().setTaskCount(3);
+ supervisor.start();
+
+ final SeekableStreamSupervisor.TaskGroup taskGroup0 =
supervisor.addTaskGroupToActivelyReadingTaskGroup(
+ supervisor.getTaskGroupIdForPartition("0"),
+ ImmutableMap.of("0", "5"),
+ null,
+ null,
+ Set.of("task0"),
+ Set.of(),
+ null
+ );
+
+ final PendingSegmentRecord record = PendingSegmentRecord.create(
+ new SegmentIdWithShardSpec("DS", Intervals.of("2024/2025"), "2024",
new NumberedShardSpec(1, 0)),
+ taskGroup0.getBaseSequenceName(),
+ "prevId",
+ "someAppendedSegment",
+ taskGroup0.getBaseSequenceName()
+ );
+
+ supervisor.registerNewVersionOfPendingSegment(record);
+
+ // The record matched task0 (so notified fires) but delivery failed over
the wire (so sendFailed fires).
+ Assert.assertEquals(1,
emitter.getMetricEventCount(SegmentUpgradeMetrics.NOTIFIED));
+ Assert.assertEquals(1,
emitter.getMetricEventCount(SegmentUpgradeMetrics.SEND_FAILED));
verifyAll();
}
diff --git a/processing/src/main/java/org/apache/druid/query/DruidMetrics.java
b/processing/src/main/java/org/apache/druid/query/DruidMetrics.java
index 532553aef6e..713ee017c00 100644
--- a/processing/src/main/java/org/apache/druid/query/DruidMetrics.java
+++ b/processing/src/main/java/org/apache/druid/query/DruidMetrics.java
@@ -58,6 +58,7 @@ public class DruidMetrics
public static final String PARTITION = "partition";
public static final String SUPERVISOR_ID = "supervisorId";
public static final String REASON = "reason";
+ public static final String VERSION = "version";
public static final String TAGS = "tags";
diff --git
a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java
b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java
index 94e7a56679b..4b1d2facc40 100644
---
a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java
+++
b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java
@@ -1181,12 +1181,53 @@ public class StreamAppenderator implements Appenderator
}
}
- public void registerUpgradedPendingSegment(PendingSegmentRecord
pendingSegmentRecord) throws IOException
+ /**
+ * Result of a {@link #registerUpgradedPendingSegment} call. Each skipped
outcome carries the human-readable
+ * {@link #getReason() reason} emitted as the {@code reason} metric
dimension.
+ */
+ public enum PendingSegmentUpgradeResult
+ {
+ ANNOUNCED(null),
+ /** The task holds no pending segment matching upgradedFromSegmentId
(request targeted the wrong task). */
+ SKIPPED_UNKNOWN_BASE("unknown base sink"),
+ /** The base sink is gone even though this task once held it. */
+ SKIPPED_NO_SINK("base sink already dropped"),
+ /** The base sink is being dropped (handoff in progress); the durable path
re-announces at the new version. */
+ SKIPPED_DROPPING("dropping base sink");
+
+ @Nullable
+ private final String reason;
+
+ PendingSegmentUpgradeResult(@Nullable String reason)
+ {
+ this.reason = reason;
+ }
+
+ /**
+ * Reason a request was skipped, for the {@code reason} metric dimension;
null for {@link #ANNOUNCED}.
+ */
+ @Nullable
+ public String getReason()
+ {
+ return reason;
+ }
+ }
+
+ public PendingSegmentUpgradeResult
registerUpgradedPendingSegment(PendingSegmentRecord pendingSegmentRecord)
throws IOException
{
SegmentIdWithShardSpec basePendingSegment =
idToPendingSegment.get(pendingSegmentRecord.getUpgradedFromSegmentId());
SegmentIdWithShardSpec upgradedPendingSegment =
pendingSegmentRecord.getId();
- if (!sinks.containsKey(basePendingSegment) ||
droppingSinks.contains(basePendingSegment)) {
- return;
+ if (basePendingSegment == null ||
droppingSinks.contains(basePendingSegment) ||
!sinks.containsKey(basePendingSegment)) {
+ if (basePendingSegment == null) {
+ // This task never allocated a segment matching upgradedFromSegmentId,
i.e. the request targeted the wrong task.
+ return PendingSegmentUpgradeResult.SKIPPED_UNKNOWN_BASE;
+ } else if (droppingSinks.contains(basePendingSegment)) {
+ // Expected during handoff: the base sink is being dropped.
+ return PendingSegmentUpgradeResult.SKIPPED_DROPPING;
+ } else {
+ // Unexpected: the base sink is gone even though this task once held
it.
+ return PendingSegmentUpgradeResult.SKIPPED_NO_SINK;
+ }
}
final Sink sink = sinks.get(basePendingSegment);
@@ -1201,6 +1242,8 @@ public class StreamAppenderator implements Appenderator
segmentAnnouncer.announceSegment(newSegment);
baseSegmentToUpgradedSegments.get(basePendingSegment).add(upgradedPendingSegment);
upgradedSegmentToBaseSegment.put(upgradedPendingSegment,
basePendingSegment);
+ log.info("Announced upgraded segment[%s] for base segment[%s] on
task[%s]", upgradedPendingSegment, basePendingSegment, myId);
+ return PendingSegmentUpgradeResult.ANNOUNCED;
}
private DataSegment getUpgradedSegment(DataSegment baseSegment,
SegmentIdWithShardSpec upgradedVersion)
diff --git
a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java
b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java
index bc5b3c6de36..4f4a0d39046 100644
---
a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java
+++
b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java
@@ -1160,6 +1160,60 @@ public class StreamAppenderatorTest extends
InitializedNullHandlingTest
}
}
+ @Test
+ public void
testRegisterUpgradedPendingSegmentReturnsAnnouncedWhenBaseSinkExists() throws
Exception
+ {
+ try (
+ final StreamAppenderatorTester tester =
+ new StreamAppenderatorTester.Builder().maxRowsInMemory(2)
+
.basePersistDirectory(temporaryFolder.newFolder())
+ .build()) {
+ final StreamAppenderator appenderator = (StreamAppenderator)
tester.getAppenderator();
+ appenderator.startJob();
+ // Create the base sink for IDENTIFIERS.get(0) so the upgrade can be
announced against it.
+ appenderator.add(IDENTIFIERS.get(0), ir("2000", "foo", 1),
Suppliers.ofInstance(Committers.nil()));
+
+ final StreamAppenderator.PendingSegmentUpgradeResult outcome =
appenderator.registerUpgradedPendingSegment(
+ PendingSegmentRecord.create(
+ si("2000/2001", "B", 1),
+ si("2000/2001", "B", 1).asSegmentId().toString(),
+ IDENTIFIERS.get(0).asSegmentId().toString(),
+ IDENTIFIERS.get(0).asSegmentId().toString(),
+ StreamAppenderatorTester.DATASOURCE
+ )
+ );
+
+
Assert.assertEquals(StreamAppenderator.PendingSegmentUpgradeResult.ANNOUNCED,
outcome);
+ }
+ }
+
+ @Test
+ public void
testRegisterUpgradedPendingSegmentReturnsSkippedUnknownBaseWhenBaseNotHeld()
throws Exception
+ {
+ try (
+ final StreamAppenderatorTester tester =
+ new StreamAppenderatorTester.Builder().maxRowsInMemory(2)
+
.basePersistDirectory(temporaryFolder.newFolder())
+ .build()) {
+ final StreamAppenderator appenderator = (StreamAppenderator)
tester.getAppenderator();
+ appenderator.startJob();
+
+ // No sink has ever been created for the upgradedFromSegmentId below, so
this task cannot announce it.
+ // This is the case where the upgrade request reached the wrong task.
+ final StreamAppenderator.PendingSegmentUpgradeResult outcome =
appenderator.registerUpgradedPendingSegment(
+ PendingSegmentRecord.create(
+ si("2050/2051", "Z", 1),
+ si("2050/2051", "Z", 1).asSegmentId().toString(),
+ si("2050/2051", "Y", 0).asSegmentId().toString(),
+ si("2050/2051", "Y", 0).asSegmentId().toString(),
+ StreamAppenderatorTester.DATASOURCE
+ )
+ );
+
+
Assert.assertEquals(StreamAppenderator.PendingSegmentUpgradeResult.SKIPPED_UNKNOWN_BASE,
outcome);
+ }
+ }
+
@Test
public void testQueryBySegments_withSegmentVersionUpgrades() throws Exception
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
