minor: Add logs and metrics observability to the segment upgrade path that fires while using concurrent locking by capistrant · Pull Request #19651 · apache/druid

capistrant · 2026-07-02T20:00:36Z

Description

Investigating potential issues with concurrent append and replace actions not being properly accepted and reflected on realtime ingestion tasks who should be upgrading their segments to reflect the concurrent append that happened on an interval they are ingesting to. This PR attempts to add some logs and metrics to the related code to help give observability into the process and smoke out if there are indeed issues going on. Biggest risk of these new mets and logs is probably chattiness, since they scale with the volume of upgrade segments that may be existing under a clusters normal operaions.

New Metrics

Metric	Description	Dimensions	Normal value
`ingest/segmentUpgrade/count`	Number of pending segments that a concurrent replace (for example, compaction) upgraded to a new version and asked the supervisor to have running tasks announce under the new version. Emitted by the replace task only when streaming ingestion is running concurrently with replace on the same interval.	`dataSource`, `taskId`, `taskType`, `groupId`, `tags`	0 unless concurrent append and replace is in use.
`ingest/segmentUpgrade/notified`	Number of upgraded pending segments the supervisor successfully routed to at least one running task. Compare with `ingest/segmentUpgrade/count`: `count` should equal `notified` + `unmatched` over the same period and `dataSource`.	`supervisorId`, `dataSource`, `stream`, `tags`	Equal to `count` in a healthy cluster.
`ingest/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`, `tags`	0. A non-zero value indicates a lost upgrade announcement.
`ingest/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`, `tags`	0
`ingest/segmentUpgrade/announced`	Number of upgraded segments a task announced under the new version. Emitted once per task, so it scales with the replica count. Do not subtract it directly from `ingest/segmentUpgrade/count`, which is per-segment rather than per-replica.	`dataSource`, `taskId`, `taskType`, `groupId`, `tags`	Greater than 0 while concurrent replace occurs.
`ingest/segmentUpgrade/skipped`	Number of upgrade requests a task received but did not announce. The `reason` dimension is one of `unknownBase` (the request reached the wrong task), `noSink` (the base sink is no longer present), or `dropping` (the base sink is handing off, which is benign and covered by the durable publish path).	`dataSource`, `taskId`, `taskType`, `groupId`, `tags`, `reason`	0, excluding `reason=dropping`.

Release note

TBD

Key changed/added classes in this PR

TBD

This PR has:

been self-reviewed.
added documentation for new or modified features or behaviors.
a release note entry in the PR description.
added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
added or updated version, license, or notice information in licenses.yaml
added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
added integration tests.
been tested in a test Druid cluster.

…ncurrent append and replace ingestion mets checkpoint doc and test tidy up self review

FrankChen021

I have reviewed the code for correctness, edge cases, concurrency/lifecycle, security, data loss, API compatibility, and missing-test risks; no high-confidence issues found.

Reviewed 11 of 11 changed files.

This is an automated review by Codex GPT-5.5

kfaraz

Thanks for adding the metrics/logs, @capistrant ! These would be instrumental in debugging failures in concurrent compaction.

Overall looks good, but I have left some nitpicks.

kfaraz · 2026-07-09T08:11:42Z

+public class SegmentUpgradeMetrics
+{
+  /** Number of upgraded pending segments a REPLACE commit created and handed to the supervisor. Task-action dims. */
+  public static final String COUNT = "ingest/segmentUpgrade/count";


Please include the prefix realtime in all the metrics to distinguish upgrade of pending segments from upgrade of appended segments.

Suggested change

public static final String COUNT = "ingest/segmentUpgrade/count";

public static final String REALTIME_PERSISTED = "ingest/realtime/segmentUpgrade/persisted";

kfaraz · 2026-07-09T08:14:04Z

+  // Values for the DruidMetrics.REASON dimension on SKIPPED.
+
+  /** The task holds no pending segment matching upgradedFromSegmentId (request targeted the wrong task). */
+  public static final String REASON_UNKNOWN_BASE = "unknownBase";


Nit: The reasons need not be camel cased, e.g. unknown base sink instead of unknownBase.

kfaraz · 2026-07-09T08:20:12Z

+  public static final String COUNT = "ingest/segmentUpgrade/count";
+
+  /** A record was delivered to at least one running task. Supervisor dims. */
+  public static final String NOTIFIED = "ingest/segmentUpgrade/notified";


Suggested change

public static final String NOTIFIED = "ingest/segmentUpgrade/notified";

public static final String REALTIME_TASK_NOTIFIED = "ingest/realtime/segmentUpgrade/notified";

kfaraz · 2026-07-09T08:48:36Z

      return;
    }

+    log.info("Registering [%d] upgraded pending segments created by task[%s] on supervisor[%s]",


Maybe move this log to the end of this method and include a summary map from pending segment ID to number of tasks notified.

kfaraz · 2026-07-09T08:49:09Z

+
+    supervisor.addTaskGroupToActivelyReadingTaskGroup(
+        supervisor.getTaskGroupIdForPartition("0"),
+        ImmutableMap.of("0", "5"),


Nit: Use Map.of and Set.of for brevity.

ack on Set. the method signature seems to require explicit ImmutableMap tho

Oh, I see. Thanks for the clarification!

kfaraz · 2026-07-09T08:50:12Z

+    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.
+        log.info(


Should be a warn, I think.

kfaraz · 2026-07-09T08:51:48Z

+        log.info(
+            "Not announcing upgraded pending segment[%s] because this task has no base sink matching"
+            + " upgradedFromSegmentId[%s]; the upgrade request likely targeted the wrong task[%s].",
+            upgradedPendingSegment, pendingSegmentRecord.getUpgradedFromSegmentId(), myId
+        );
+        return UpgradeAnnouncementOutcome.SKIPPED_UNKNOWN_BASE;
+      } else if (droppingSinks.contains(basePendingSegment)) {
+        // Expected during handoff: the base sink is being dropped
+        log.debug(
+            "Not announcing upgraded pending segment[%s] for base segment[%s] on task[%s] because the base sink is being dropped.",
+            upgradedPendingSegment, basePendingSegment, myId
+        );
+        return UpgradeAnnouncementOutcome.SKIPPED_DROPPING;
+      } else {
+        // Unexpected: the base sink is gone even though this task once held it.
+        log.info(
+            "Not announcing upgraded pending segment[%s] for base segment[%s] on task[%s] because the base sink is no longer present.",
+            upgradedPendingSegment, basePendingSegment, myId


I feel like these logs can be merged into one and can be logged by the caller itself using the result of this method.

applying the pure text changes. will commit responses to review in another commit Co-authored-by: Kashif Faraz <kashif.faraz@gmail.com>

kfaraz · 2026-07-10T06:49:44Z

 |`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/count`|Number of pending segments that a concurrent replace (for example, compaction) upgraded to a new version and asked the supervisor to have running tasks announce under the new version. Emitted by the replace task only when streaming ingestion is running concurrently with replace on the same interval.|`dataSource`, `taskId`, `taskType`, `groupId`, `interval`, `version`, `tags`|0 unless [concurrent append and replace](../ingestion/concurrent-append-replace.md) is in use.|


We should probably rename this metric from count to persisted to signify that this is the count that was actually persisted to the metadata store.
Also, rephrased the text a bit.

Suggested change

|`ingest/realtime/segmentUpgrade/count`|Number of pending segments that a concurrent replace (for example, compaction) upgraded to a new version and asked the supervisor to have running tasks announce under the new version. Emitted by the replace task only when streaming ingestion is running concurrently with replace on the same interval.|`dataSource`, `taskId`, `taskType`, `groupId`, `interval`, `version`, `tags`|0 unless [concurrent append and replace](../ingestion/concurrent-append-replace.md) is in use.|

|`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`, `version`, `tags`|0 unless [concurrent append and replace](../ingestion/concurrent-append-replace.md) is in use.|

kfaraz · 2026-07-10T06:51:42Z

+   * 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 setSegmentDimensions(


Nit: Maybe move this method to IndexTaskUtils itself, as it feels like a more generic place (rather than upgrade specific) allowing for reuse for this method.

kfaraz · 2026-07-10T06:55:09Z

+      log.info("No active streaming supervisor for datasource[%s]; the [%d] upgraded pending segment(s) from task[%s]"
+               + " will become queryable when their tasks hand off.",
+                task.getDataSource(),
+                upgradedPendingSegments.size(),
+                task.getId()
+      );


minor rephrase

Suggested change

log.info("No active streaming supervisor for datasource[%s]; the [%d] upgraded pending segment(s) from task[%s]"

+ " will become queryable when their tasks hand off.",

task.getDataSource(),

upgradedPendingSegments.size(),

task.getId()

);

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()

);

kfaraz · 2026-07-10T07:00:17Z

+  /**
+   * Notifies every running task in the matching task group(s) of an upgraded pending segment.
+   *
+   * @return the number of tasks notified, which the caller aggregates into a per-batch summary


Suggested change

* @return the number of tasks notified, which the caller aggregates into a per-batch summary

* @return the number of tasks notified

kfaraz · 2026-07-10T07:01:13Z

    }
+
+    // Only the exceptional no-match case is logged per segment; the notified count is returned to the task action,
+    // which summarizes the whole batch in one log line, and each notification is captured by the per-task metric.


We need not call this out. This method should be agnostic of the behaviour of its caller.
We choose not to log the successful case here to reduce verbosity, irrespective of the behaviour of the caller.

kfaraz · 2026-07-10T07:03:23Z

+          "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. Currently tracking [%d] activelyReading"
+          + " and [%d] pendingCompletion task group(s).",


Do you think logging the number of actively reading / pending completion groups would be useful for debugging? I feel we may omit these for now.

hmm I guess probably not useful. was kinda just packing in bits of information wherever I could find them in this PR. but you're right I'm not sure what I do with this number in this log so I will remove

kfaraz · 2026-07-10T07:05:23Z

+  {
+    ANNOUNCED(null),
+    /** The task holds no pending segment matching upgradedFromSegmentId (request targeted the wrong task). */
+    SKIPPED_UNKNOWN_BASE("unknown base"),


Suggested change

SKIPPED_UNKNOWN_BASE("unknown base"),

SKIPPED_UNKNOWN_BASE("unknown base sink"),

kfaraz · 2026-07-10T07:07:02Z

+   * Logs and emits a metric for the outcome of a pending-segment upgrade request, keyed off the {@code outcome}
+   * returned by {@link StreamAppenderator#registerUpgradedPendingSegment}.
+   */
+  private void recordUpgradeOutcome(


Suggested change

private void recordUpgradeOutcome(

private void recordUpgradeResult(

kfaraz · 2026-07-10T07:07:10Z

+   */
+  private void recordUpgradeOutcome(
+      PendingSegmentRecord upgradedPendingSegment,
+      StreamAppenderator.PendingSegmentUpgradeResult outcome


Suggested change

StreamAppenderator.PendingSegmentUpgradeResult outcome

StreamAppenderator.PendingSegmentUpgradeResult result

FrankChen021

Severity	Findings
P0	0
P1	1
P2	1
P3	0
Total	2

Severity	Findings
P0	0
P1	1
P2	1
P3	0
Total	2

Reviewed 16 of 16 changed files.

This is an automated review by Codex GPT-5.6-Sol

FrankChen021 · 2026-07-10T12:03:48Z

  "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/count" : { "dimensions" : ["dataSource", "interval", "version"], "type" : "count", "help": "Number of pending segments upgraded by a concurrent replace." },


[P1] Avoid unbounded per-version Prometheus series

The default exporter has no TTL unless flushPeriod is explicitly configured, and its collector retains every label tuple. version, interval, and taskId continually change across replacements and tasks, so recurring concurrent compaction makes these six default metrics accumulate series indefinitely on long-lived Overlord/supervisor processes, potentially exhausting heap. Keep high-cardinality identifiers out of the default mapping or enforce bounded expiry.

FrankChen021 · 2026-07-10T12:03:48Z

+        notifiedTasksBySegment.size(),
+        task.getId(),
+        activeSupervisorIdWithAppendLock.get(),
+        notifiedTasksBySegment


[P2] Bound the per-segment INFO summary

The upgraded-segment list is not capped, yet this builds and formats every segment ID into one INFO message. Broad compaction can produce thousands of entries and megabyte-scale log lines on the task-action path. Log aggregate counts with a bounded sample, reserving full detail for debug output.

Add logs and metrics observability to the segment upgrade path for co…

58381a5

…ncurrent append and replace ingestion mets checkpoint doc and test tidy up self review

github-actions Bot added Area - Documentation Area - Metrics/Event Emitting Area - Ingestion labels Jul 2, 2026

FrankChen021 reviewed Jul 3, 2026

View reviewed changes

kfaraz reviewed Jul 9, 2026

View reviewed changes

capistrant and others added 2 commits July 9, 2026 20:32

Apply suggestions from code review

d82041f

applying the pure text changes. will commit responses to review in another commit Co-authored-by: Kashif Faraz <kashif.faraz@gmail.com>

upgrade metrics names

1999407

github-advanced-security AI found potential problems Jul 10, 2026

View reviewed changes

Comment thread .../main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java Fixed

capistrant added 4 commits July 9, 2026 22:00

review responses

b2c75e2

fixup statsd emitter json

dbe7b65

dropwizard json def update

1a83b18

detailed task action logging

03c48a2

kfaraz approved these changes Jul 10, 2026

View reviewed changes

kfaraz mentioned this pull request Jul 10, 2026

minor: Add segment/allocated/count metric #19674

Open

10 tasks

FrankChen021 reviewed Jul 10, 2026

View reviewed changes

capistrant added 2 commits July 10, 2026 09:58

more review comments

ad29253

coverage

1a3c899

	public static final String COUNT = "ingest/segmentUpgrade/count";
	public static final String REALTIME_PERSISTED = "ingest/realtime/segmentUpgrade/persisted";

	public static final String NOTIFIED = "ingest/segmentUpgrade/notified";
	public static final String REALTIME_TASK_NOTIFIED = "ingest/realtime/segmentUpgrade/notified";

	* @return the number of tasks notified, which the caller aggregates into a per-batch summary
	* @return the number of tasks notified

	SKIPPED_UNKNOWN_BASE("unknown base"),
	SKIPPED_UNKNOWN_BASE("unknown base sink"),

	private void recordUpgradeOutcome(
	private void recordUpgradeResult(

	StreamAppenderator.PendingSegmentUpgradeResult outcome
	StreamAppenderator.PendingSegmentUpgradeResult result

Uh oh!

Conversation

capistrant commented Jul 2, 2026

Description

New Metrics

Release note

Key changed/added classes in this PR

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

kfaraz left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

kfaraz Jul 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kfaraz Jul 9, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

FrankChen021 left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

kfaraz Jul 9, 2026 •

edited

Loading

kfaraz Jul 9, 2026 •

edited

Loading