kfaraz commented on code in PR #18561:
URL: https://github.com/apache/druid/pull/18561#discussion_r2371109597
##########
sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java:
##########
@@ -277,6 +290,21 @@ public void refresh(final Set<SegmentId>
segmentsToRefresh, final Set<String> da
}
}
+ private Map<String, Integer> getDatasourceToSegmentsSkipped(
+ Set<SegmentId> segmentsToRefresh,
+ Map<String, PhysicalDatasourceMetadata> polledDataSourceMetadata
+ )
+ {
+ final Map<String, Integer> datasourceToNumSegmentsSkipped = new
HashMap<>();
+
+ for (SegmentId segmentId : segmentsToRefresh) {
+ if (polledDataSourceMetadata.containsKey(segmentId.getDataSource())) {
+ datasourceToNumSegmentsSkipped.merge(segmentId.getDataSource(), 1,
Integer::sum);
+ }
+ }
+ return datasourceToNumSegmentsSkipped;
Review Comment:
Don't return anything. Perform the metric emission here itself.
##########
sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java:
##########
@@ -223,8 +223,21 @@ public void refresh(final Set<SegmentId>
segmentsToRefresh, final Set<String> da
polledDataSourceMetadata.forEach(this::updateDSMetadata);
// Remove segments of the datasource from refresh list for which we
received schema from the Coordinator.
+ final Map<String, Integer> datasourceToNumSegmentsSkipped =
getDatasourceToSegmentsSkipped(segmentsToRefresh, polledDataSourceMetadata);
+
segmentsToRefresh.removeIf(segmentId ->
polledDataSourceMetadata.containsKey(segmentId.getDataSource()));
+ // Emit metrics per datasource
+ datasourceToNumSegmentsSkipped.forEach(
+ (dataSource, count) ->
+ emitMetric(
+ Metric.BROKER_SEGMENTS_SKIPPED_REFRESH,
+ count,
+ new ServiceMetricEvent.Builder().setDimension(
+ DruidMetrics.DATASOURCE,
+ dataSource))
+ );
Review Comment:
This should be put in the new method as well.
##########
sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java:
##########
@@ -223,8 +223,21 @@ public void refresh(final Set<SegmentId>
segmentsToRefresh, final Set<String> da
polledDataSourceMetadata.forEach(this::updateDSMetadata);
// Remove segments of the datasource from refresh list for which we
received schema from the Coordinator.
+ final Map<String, Integer> datasourceToNumSegmentsSkipped =
getDatasourceToSegmentsSkipped(segmentsToRefresh, polledDataSourceMetadata);
+
segmentsToRefresh.removeIf(segmentId ->
polledDataSourceMetadata.containsKey(segmentId.getDataSource()));
+ // Emit metrics per datasource
+ datasourceToNumSegmentsSkipped.forEach(
+ (dataSource, count) ->
+ emitMetric(
+ Metric.BROKER_SEGMENTS_SKIPPED_REFRESH,
+ count,
+ new ServiceMetricEvent.Builder().setDimension(
+ DruidMetrics.DATASOURCE,
+ dataSource))
Review Comment:
```suggestion
new
ServiceMetricEvent.Builder().setDimension(DruidMetrics.DATASOURCE, dataSource)
)
```
##########
sql/src/main/java/org/apache/druid/sql/calcite/schema/BrokerSegmentMetadataCache.java:
##########
@@ -277,6 +290,21 @@ public void refresh(final Set<SegmentId>
segmentsToRefresh, final Set<String> da
}
}
+ private Map<String, Integer> getDatasourceToSegmentsSkipped(
Review Comment:
```suggestion
private void emitMetricForSkippedSegments(
```
--
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]
