kfaraz commented on code in PR #19651: URL: https://github.com/apache/druid/pull/19651#discussion_r3550009811
########## 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 #COUNT} (how many upgrades a commit produced),</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> + * Comparing {@link #COUNT} (created) against {@link #ANNOUNCED} (applied) per dataSource quantifies the visibility gap; + * {@link #UNMATCHED}, {@link #SEND_FAILED} and {@link #SKIPPED} attribute where a lost upgrade dropped out. + */ +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"; + + /** A record was delivered to at least one running task. Supervisor dims. */ + public static final String NOTIFIED = "ingest/segmentUpgrade/notified"; + + /** A record matched no running task and will not be re-announced until handoff. Supervisor dims. */ + public static final String UNMATCHED = "ingest/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/segmentUpgrade/sendFailed"; + + /** A task announced an upgraded segment under the new version. Task dims. */ + public static final String ANNOUNCED = "ingest/segmentUpgrade/announced"; + + /** A task received an upgrade request but did not announce it; see the {@code reason} dimension. Task dims. */ + public static final String SKIPPED = "ingest/segmentUpgrade/skipped"; + + // 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"; Review Comment: Nit: The reasons need not be camel cased, e.g. `unknown base sink` instead of `unknownBase`. ########## 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 #COUNT} (how many upgrades a commit produced),</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> + * Comparing {@link #COUNT} (created) against {@link #ANNOUNCED} (applied) per dataSource quantifies the visibility gap; + * {@link #UNMATCHED}, {@link #SEND_FAILED} and {@link #SKIPPED} attribute where a lost upgrade dropped out. + */ +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"; Review Comment: Please include the prefix `realtime` in all the metrics to distinguish upgrade of pending segments from upgrade of appended segments. ```suggestion public static final String REALTIME_PERSISTED = "ingest/realtime/segmentUpgrade/persisted"; ``` ########## 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 #COUNT} (how many upgrades a commit produced),</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> + * Comparing {@link #COUNT} (created) against {@link #ANNOUNCED} (applied) per dataSource quantifies the visibility gap; + * {@link #UNMATCHED}, {@link #SEND_FAILED} and {@link #SKIPPED} attribute where a lost upgrade dropped out. + */ +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"; + + /** A record was delivered to at least one running task. Supervisor dims. */ + public static final String NOTIFIED = "ingest/segmentUpgrade/notified"; + + /** A record matched no running task and will not be re-announced until handoff. Supervisor dims. */ + public static final String UNMATCHED = "ingest/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/segmentUpgrade/sendFailed"; + + /** A task announced an upgraded segment under the new version. Task dims. */ + public static final String ANNOUNCED = "ingest/segmentUpgrade/announced"; + + /** A task received an upgrade request but did not announce it; see the {@code reason} dimension. Task dims. */ + public static final String SKIPPED = "ingest/segmentUpgrade/skipped"; + + // 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"; + /** The base sink is gone even though this task once held it. */ + public static final String REASON_NO_SINK = "noSink"; + /** The base sink is being dropped (handoff in progress); the durable path re-announces at the new version. */ + public static final String REASON_DROPPING = "dropping"; Review Comment: ```suggestion public static final String REASON_DROPPING = "dropping base sink"; ``` ########## 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 #COUNT} (how many upgrades a commit produced),</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> + * Comparing {@link #COUNT} (created) against {@link #ANNOUNCED} (applied) per dataSource quantifies the visibility gap; + * {@link #UNMATCHED}, {@link #SEND_FAILED} and {@link #SKIPPED} attribute where a lost upgrade dropped out. + */ +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"; + + /** A record was delivered to at least one running task. Supervisor dims. */ + public static final String NOTIFIED = "ingest/segmentUpgrade/notified"; + + /** A record matched no running task and will not be re-announced until handoff. Supervisor dims. */ + public static final String UNMATCHED = "ingest/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/segmentUpgrade/sendFailed"; + + /** A task announced an upgraded segment under the new version. Task dims. */ + public static final String ANNOUNCED = "ingest/segmentUpgrade/announced"; + + /** A task received an upgrade request but did not announce it; see the {@code reason} dimension. Task dims. */ + public static final String SKIPPED = "ingest/segmentUpgrade/skipped"; + + // 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"; + /** The base sink is gone even though this task once held it. */ + public static final String REASON_NO_SINK = "noSink"; Review Comment: ```suggestion public static final String REASON_NO_SINK = "base sink already dropped"; ``` ########## indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java: ########## @@ -1419,22 +1421,89 @@ public void registerNewVersionOfPendingSegment( PendingSegmentRecord pendingSegmentRecord ) { + final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId(); + int matchedGroups = 0; + int notifiedTasks = 0; + for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) { - if (taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) { + if (taskGroup.baseSequenceName.equals(taskAllocatorId)) { + matchedGroups++; 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)) { + matchedGroups++; 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( + "Upgraded pending segment[%s] (upgradedFrom[%s], taskAllocatorId[%s]) matched no running task on" + + " supervisor[%s]; it will not be re-announced until handoff. Currently tracking [%d] activelyReading" + + " and [%d] pendingCompletion task group(s).", + pendingSegmentRecord.getId(), + pendingSegmentRecord.getUpgradedFromSegmentId(), + taskAllocatorId, + supervisorId, + activelyReadingTaskGroups.size(), + pendingCompletionTaskGroups.size() + ); + emitter.emit(getMetricBuilder().setMetric(SegmentUpgradeMetrics.UNMATCHED, 1)); + } else { + log.info( Review Comment: Let's skip logging this for every pending segment that has been upgraded. We can just log a total count summary in the task action. ########## 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 #COUNT} (how many upgrades a commit produced),</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> + * Comparing {@link #COUNT} (created) against {@link #ANNOUNCED} (applied) per dataSource quantifies the visibility gap; + * {@link #UNMATCHED}, {@link #SEND_FAILED} and {@link #SKIPPED} attribute where a lost upgrade dropped out. + */ +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"; + + /** A record was delivered to at least one running task. Supervisor dims. */ + public static final String NOTIFIED = "ingest/segmentUpgrade/notified"; Review Comment: ```suggestion public static final String REALTIME_TASK_NOTIFIED = "ingest/segmentUpgrade/notified"; ``` ########## indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java: ########## @@ -1419,22 +1421,89 @@ public void registerNewVersionOfPendingSegment( PendingSegmentRecord pendingSegmentRecord ) { + final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId(); + int matchedGroups = 0; + int notifiedTasks = 0; + for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) { - if (taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) { + if (taskGroup.baseSequenceName.equals(taskAllocatorId)) { + matchedGroups++; 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)) { + matchedGroups++; 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( + "Upgraded pending segment[%s] (upgradedFrom[%s], taskAllocatorId[%s]) matched no running task on" Review Comment: Nit: slight rephrase ```suggestion "Could not find any task matching taskAllocatorId[%s] in supervisor[%s] for upgraded pending segment[%s] (upgradedFrom[%s]) " ``` ########## 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 #COUNT} (how many upgrades a commit produced),</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> + * Comparing {@link #COUNT} (created) against {@link #ANNOUNCED} (applied) per dataSource quantifies the visibility gap; + * {@link #UNMATCHED}, {@link #SEND_FAILED} and {@link #SKIPPED} attribute where a lost upgrade dropped out. + */ +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"; + + /** A record was delivered to at least one running task. Supervisor dims. */ + public static final String NOTIFIED = "ingest/segmentUpgrade/notified"; + + /** A record matched no running task and will not be re-announced until handoff. Supervisor dims. */ + public static final String UNMATCHED = "ingest/segmentUpgrade/unmatched"; Review Comment: I wonder if we shouldn't just club this metric with `ingest/realtime/segmentUpgrade/skipped` and add a reason "no matching task" ########## indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java: ########## @@ -1419,22 +1421,89 @@ public void registerNewVersionOfPendingSegment( PendingSegmentRecord pendingSegmentRecord ) { + final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId(); + int matchedGroups = 0; + int notifiedTasks = 0; + for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) { - if (taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) { + if (taskGroup.baseSequenceName.equals(taskAllocatorId)) { + matchedGroups++; 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)) { + matchedGroups++; 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( + "Upgraded pending segment[%s] (upgradedFrom[%s], taskAllocatorId[%s]) matched no running task on" + + " supervisor[%s]; it will not be re-announced until handoff. Currently tracking [%d] activelyReading" + + " and [%d] pendingCompletion task group(s).", + pendingSegmentRecord.getId(), + pendingSegmentRecord.getUpgradedFromSegmentId(), + taskAllocatorId, + supervisorId, + activelyReadingTaskGroups.size(), + pendingCompletionTaskGroups.size() + ); + emitter.emit(getMetricBuilder().setMetric(SegmentUpgradeMetrics.UNMATCHED, 1)); Review Comment: Let's include the segment version and interval in all the metrics. ########## indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java: ########## @@ -1419,22 +1421,89 @@ public void registerNewVersionOfPendingSegment( PendingSegmentRecord pendingSegmentRecord ) { + final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId(); + int matchedGroups = 0; + int notifiedTasks = 0; + for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) { - if (taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) { + if (taskGroup.baseSequenceName.equals(taskAllocatorId)) { + matchedGroups++; 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)) { + matchedGroups++; 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( + "Upgraded pending segment[%s] (upgradedFrom[%s], taskAllocatorId[%s]) matched no running task on" + + " supervisor[%s]; it will not be re-announced until handoff. Currently tracking [%d] activelyReading" + + " and [%d] pendingCompletion task group(s).", + pendingSegmentRecord.getId(), + pendingSegmentRecord.getUpgradedFromSegmentId(), + taskAllocatorId, + supervisorId, + activelyReadingTaskGroups.size(), + pendingCompletionTaskGroups.size() + ); + emitter.emit(getMetricBuilder().setMetric(SegmentUpgradeMetrics.UNMATCHED, 1)); + } else { + log.info( + "Notified [%d] task(s) across [%d] task group(s) of upgraded pending segment[%s] (upgradedFrom[%s]).", + notifiedTasks, + matchedGroups, + pendingSegmentRecord.getId(), + pendingSegmentRecord.getUpgradedFromSegmentId() + ); + emitter.emit(getMetricBuilder().setMetric(SegmentUpgradeMetrics.NOTIFIED, 1)); + } + } + + /** + * Sends an upgraded pending segment to a single task and records a metric if the request fails to reach the task. + */ + private void notifyTaskOfUpgradedPendingSegment(String taskId, PendingSegmentRecord pendingSegmentRecord) + { + Futures.addCallback( + taskClient.registerNewVersionOfPendingSegmentAsync(taskId, pendingSegmentRecord), + new FutureCallback<>() + { + @Override + public void onSuccess(Boolean result) + { + // Successful delivery is captured by the notified metric; nothing to do here. Review Comment: Nit: I don't think this is entirely true since the `notified` metric only captures the fact that we have sent a notification, not that it has been acknowledged successfully. But I agree that we can skip emitting a metric here. (since the lack of a `sendFailed` metric should suffice). ########## indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java: ########## @@ -1419,22 +1421,89 @@ public void registerNewVersionOfPendingSegment( PendingSegmentRecord pendingSegmentRecord ) { + final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId(); + int matchedGroups = 0; + int notifiedTasks = 0; + for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) { - if (taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) { + if (taskGroup.baseSequenceName.equals(taskAllocatorId)) { + matchedGroups++; 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)) { + matchedGroups++; 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( + "Upgraded pending segment[%s] (upgradedFrom[%s], taskAllocatorId[%s]) matched no running task on" + + " supervisor[%s]; it will not be re-announced until handoff. Currently tracking [%d] activelyReading" + + " and [%d] pendingCompletion task group(s).", + pendingSegmentRecord.getId(), + pendingSegmentRecord.getUpgradedFromSegmentId(), + taskAllocatorId, + supervisorId, + activelyReadingTaskGroups.size(), + pendingCompletionTaskGroups.size() + ); + emitter.emit(getMetricBuilder().setMetric(SegmentUpgradeMetrics.UNMATCHED, 1)); + } else { + log.info( + "Notified [%d] task(s) across [%d] task group(s) of upgraded pending segment[%s] (upgradedFrom[%s]).", + notifiedTasks, + matchedGroups, + pendingSegmentRecord.getId(), + pendingSegmentRecord.getUpgradedFromSegmentId() + ); + emitter.emit(getMetricBuilder().setMetric(SegmentUpgradeMetrics.NOTIFIED, 1)); + } + } + + /** + * Sends an upgraded pending segment to a single task and records a metric if the request fails to reach the task. + */ + private void notifyTaskOfUpgradedPendingSegment(String taskId, PendingSegmentRecord pendingSegmentRecord) + { + Futures.addCallback( + taskClient.registerNewVersionOfPendingSegmentAsync(taskId, pendingSegmentRecord), + new FutureCallback<>() + { + @Override + public void onSuccess(Boolean result) + { + // Successful delivery is captured by the notified metric; nothing to do here. + } + + @Override + public void onFailure(Throwable t) + { + log.warn( + t, + "Failed to send upgraded pending segment[%s] to task[%s] on supervisor[%s].", Review Comment: ```suggestion "Failed to register upgraded pending segment[%s] on task[%s] of supervisor[%s].", ``` ########## indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java: ########## @@ -1419,22 +1421,89 @@ public void registerNewVersionOfPendingSegment( PendingSegmentRecord pendingSegmentRecord ) { + final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId(); + int matchedGroups = 0; + int notifiedTasks = 0; + for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) { - if (taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) { + if (taskGroup.baseSequenceName.equals(taskAllocatorId)) { + matchedGroups++; 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)) { + matchedGroups++; 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( + "Upgraded pending segment[%s] (upgradedFrom[%s], taskAllocatorId[%s]) matched no running task on" + + " supervisor[%s]; it will not be re-announced until handoff. Currently tracking [%d] activelyReading" + + " and [%d] pendingCompletion task group(s).", + pendingSegmentRecord.getId(), + pendingSegmentRecord.getUpgradedFromSegmentId(), + taskAllocatorId, + supervisorId, + activelyReadingTaskGroups.size(), + pendingCompletionTaskGroups.size() + ); + emitter.emit(getMetricBuilder().setMetric(SegmentUpgradeMetrics.UNMATCHED, 1)); + } else { + log.info( + "Notified [%d] task(s) across [%d] task group(s) of upgraded pending segment[%s] (upgradedFrom[%s]).", + notifiedTasks, + matchedGroups, + pendingSegmentRecord.getId(), + pendingSegmentRecord.getUpgradedFromSegmentId() + ); + emitter.emit(getMetricBuilder().setMetric(SegmentUpgradeMetrics.NOTIFIED, 1)); Review Comment: Let's emit the number of tasks notified instead of always emitting 1. (I wonder if we should also emit the `taskId` that was notified) ########## server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java: ########## @@ -1181,12 +1181,46 @@ private void unannounceSegment(DataSegment segment) } } - public void registerUpgradedPendingSegment(PendingSegmentRecord pendingSegmentRecord) throws IOException + /** + * Outcome of a {@link #registerUpgradedPendingSegment} call, returned so the caller (which owns the emitter) can + * emit the corresponding metric. + */ + public enum UpgradeAnnouncementOutcome Review Comment: ```suggestion public enum PendingSegmentUpgradeResult ``` ########## indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java: ########## @@ -169,14 +171,33 @@ private void registerUpgradedPendingSegmentsOnSupervisor( List<PendingSegmentRecord> upgradedPendingSegments ) { + // Emit the count of upgrades this commit produced regardless of whether a supervisor exists to Review Comment: ```suggestion // Emit the count of upgrades successfully committed regardless of whether a supervisor exists to ``` ########## indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java: ########## @@ -2116,6 +2117,85 @@ public void testRegisterNewVersionOfPendingSegment() 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"), Review Comment: Nit: Use `Map.of` and `Set.of` for brevity. ########## indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java: ########## @@ -169,14 +171,33 @@ private void registerUpgradedPendingSegmentsOnSupervisor( List<PendingSegmentRecord> upgradedPendingSegments ) { + // Emit the count of upgrades this commit produced regardless of whether a supervisor exists to + // receive them, so it can be compared against the count actually announced by tasks. + if (!upgradedPendingSegments.isEmpty()) { + final ServiceMetricEvent.Builder metricBuilder = new ServiceMetricEvent.Builder(); + IndexTaskUtils.setTaskDimensions(metricBuilder, task); + toolbox.getEmitter().emit(metricBuilder.setMetric(SegmentUpgradeMetrics.COUNT, upgradedPendingSegments.size())); + } + final SupervisorManager supervisorManager = toolbox.getSupervisorManager(); final Optional<String> activeSupervisorIdWithAppendLock = supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(task.getDataSource()); if (!activeSupervisorIdWithAppendLock.isPresent()) { + 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() + ); return; } + log.info("Registering [%d] upgraded pending segments created by task[%s] on supervisor[%s]", Review Comment: Nit: for better formatting, start the message in a new line. ```suggestion log.info( "Registering [%d] upgraded pending segments created by task[%s] on supervisor[%s]", ``` ########## indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java: ########## @@ -169,14 +171,33 @@ private void registerUpgradedPendingSegmentsOnSupervisor( List<PendingSegmentRecord> upgradedPendingSegments ) { + // Emit the count of upgrades this commit produced regardless of whether a supervisor exists to + // receive them, so it can be compared against the count actually announced by tasks. + if (!upgradedPendingSegments.isEmpty()) { + final ServiceMetricEvent.Builder metricBuilder = new ServiceMetricEvent.Builder(); + IndexTaskUtils.setTaskDimensions(metricBuilder, task); + toolbox.getEmitter().emit(metricBuilder.setMetric(SegmentUpgradeMetrics.COUNT, upgradedPendingSegments.size())); + } + final SupervisorManager supervisorManager = toolbox.getSupervisorManager(); final Optional<String> activeSupervisorIdWithAppendLock = supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(task.getDataSource()); if (!activeSupervisorIdWithAppendLock.isPresent()) { + 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() + ); return; } + log.info("Registering [%d] upgraded pending segments created by task[%s] on supervisor[%s]", + upgradedPendingSegments.size(), + task.getId(), + activeSupervisorIdWithAppendLock.get() + ); upgradedPendingSegments.forEach( upgradedPendingSegment -> supervisorManager.registerUpgradedPendingSegmentOnSupervisor( Review Comment: Maybe this method should return the number of tasks that were notified for this pending segment. We could log a summary map of pending segment ID to number of tasks notified. ########## server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java: ########## @@ -1181,12 +1181,46 @@ private void unannounceSegment(DataSegment segment) } } - public void registerUpgradedPendingSegment(PendingSegmentRecord pendingSegmentRecord) throws IOException + /** + * Outcome of a {@link #registerUpgradedPendingSegment} call, returned so the caller (which owns the emitter) can + * emit the corresponding metric. Review Comment: ```suggestion * Result of a {@link #registerUpgradedPendingSegment} call. ``` ########## server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java: ########## @@ -1181,12 +1181,46 @@ private void unannounceSegment(DataSegment segment) } } - public void registerUpgradedPendingSegment(PendingSegmentRecord pendingSegmentRecord) throws IOException + /** + * Outcome of a {@link #registerUpgradedPendingSegment} call, returned so the caller (which owns the emitter) can + * emit the corresponding metric. + */ + public enum UpgradeAnnouncementOutcome + { + ANNOUNCED, + SKIPPED_UNKNOWN_BASE, + SKIPPED_NO_SINK, + SKIPPED_DROPPING + } + + public UpgradeAnnouncementOutcome 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. + log.info( Review Comment: Should be a warn, I think. ########## server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java: ########## @@ -1181,12 +1181,46 @@ private void unannounceSegment(DataSegment segment) } } - public void registerUpgradedPendingSegment(PendingSegmentRecord pendingSegmentRecord) throws IOException + /** + * Outcome of a {@link #registerUpgradedPendingSegment} call, returned so the caller (which owns the emitter) can + * emit the corresponding metric. + */ + public enum UpgradeAnnouncementOutcome + { + ANNOUNCED, + SKIPPED_UNKNOWN_BASE, + SKIPPED_NO_SINK, + SKIPPED_DROPPING + } + + public UpgradeAnnouncementOutcome 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. + 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 Review Comment: I feel like these logs can be merged into and can be logged by the caller itself using the result of this method. ########## indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java: ########## @@ -1865,6 +1869,33 @@ public Response registerUpgradedPendingSegment( } } + private void emitUpgradeAnnouncementMetric(StreamAppenderator.UpgradeAnnouncementOutcome outcome) + { + if (outcome == StreamAppenderator.UpgradeAnnouncementOutcome.ANNOUNCED) { + task.emitMetric(toolbox.getEmitter(), SegmentUpgradeMetrics.ANNOUNCED, 1); + return; + } + final String reason; + switch (outcome) { + case SKIPPED_UNKNOWN_BASE: + reason = SegmentUpgradeMetrics.REASON_UNKNOWN_BASE; + break; + case SKIPPED_NO_SINK: + reason = SegmentUpgradeMetrics.REASON_NO_SINK; + break; + case SKIPPED_DROPPING: + reason = SegmentUpgradeMetrics.REASON_DROPPING; + break; Review Comment: Can we capture the full reason string inside the enum itself? ########## indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java: ########## @@ -169,14 +171,33 @@ private void registerUpgradedPendingSegmentsOnSupervisor( List<PendingSegmentRecord> upgradedPendingSegments ) { + // Emit the count of upgrades this commit produced regardless of whether a supervisor exists to + // receive them, so it can be compared against the count actually announced by tasks. + if (!upgradedPendingSegments.isEmpty()) { + final ServiceMetricEvent.Builder metricBuilder = new ServiceMetricEvent.Builder(); + IndexTaskUtils.setTaskDimensions(metricBuilder, task); + toolbox.getEmitter().emit(metricBuilder.setMetric(SegmentUpgradeMetrics.COUNT, upgradedPendingSegments.size())); + } + final SupervisorManager supervisorManager = toolbox.getSupervisorManager(); final Optional<String> activeSupervisorIdWithAppendLock = supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(task.getDataSource()); if (!activeSupervisorIdWithAppendLock.isPresent()) { + 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() + ); return; } + log.info("Registering [%d] upgraded pending segments created by task[%s] on supervisor[%s]", Review Comment: Maybe move this log to the end of this method and include a summary map from pending segment ID to number of tasks notified. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
