Re-enable centalized schema with no query embedded tests#18561
Merged
kfaraz merged 3 commits intoapache:masterfrom Sep 23, 2025
Merged
Re-enable centalized schema with no query embedded tests#18561kfaraz merged 3 commits intoapache:masterfrom
kfaraz merged 3 commits intoapache:masterfrom
Conversation
kfaraz
reviewed
Sep 23, 2025
| * to be queryable by Brokers. | ||
| */ | ||
| public void waitForAllSegmentsToBeAvailable(String dataSource, EmbeddedCoordinator coordinator, EmbeddedBroker broker) | ||
| public void waitForAllSegmentsToBeAvailable( |
Contributor
There was a problem hiding this comment.
We shouldn't change this method. Please add a new method which waits on the new metric.
| final String taskId = IdUtils.getRandomId(); | ||
| cluster.callApi().runTask(taskBuilder.dataSource(dataSource).withId(taskId), overlord); | ||
| cluster.callApi().waitForAllSegmentsToBeAvailable(dataSource, coordinator, broker); | ||
| boolean useCentralizedSchema = Boolean.parseBoolean((String) cluster.getCommonProperties().getOrDefault("druid.centralizedDatasourceSchema.enabled", "false")); |
Contributor
There was a problem hiding this comment.
Suggested change
| boolean useCentralizedSchema = Boolean.parseBoolean((String) cluster.getCommonProperties().getOrDefault("druid.centralizedDatasourceSchema.enabled", "false")); | |
| boolean useCentralizedSchema = Boolean.parseBoolean(cluster.getCommonProperties().getProperty("druid.centralizedDatasourceSchema.enabled", "false")); |
| .addCommonProperty("druid.centralizedDatasourceSchema.backFillEnabled", "true") | ||
| .addCommonProperty("druid.centralizedDatasourceSchema.backFillPeriod", "500") | ||
| .addCommonProperty("druid.coordinator.segmentMetadata.disableSegmentMetadataQueries", "true") | ||
| .addCommonProperty("druid.sql.planner.metadataRefreshPeriod", "PT0.1s") |
Contributor
There was a problem hiding this comment.
This property is already added to EmbeddedBroker. Do we need to add it here too?
| polledDataSourceMetadata.forEach(this::updateDSMetadata); | ||
|
|
||
| // Remove segments of the datasource from refresh list for which we received schema from the Coordinator. | ||
| final Map<String, Integer> segmentsSkippedPerDatasource = new HashMap<>(); |
Contributor
There was a problem hiding this comment.
Suggested change
| final Map<String, Integer> segmentsSkippedPerDatasource = new HashMap<>(); | |
| final Map<String, Integer> datasourceToNumSegmentsSkipped = new HashMap<>(); |
Comment on lines
+239
to
+240
| emitMetric(Metric.BROKER_SEGMENTS_SKIPPED_REFRESH, count, | ||
| new ServiceMetricEvent.Builder().setDimension(DruidMetrics.DATASOURCE, dataSource)); |
Contributor
There was a problem hiding this comment.
Please put each arg in a separate line.
| segmentsToRefresh.removeIf(segmentId -> polledDataSourceMetadata.containsKey(segmentId.getDataSource())); | ||
|
|
||
| // Emit metrics per datasource | ||
| segmentsSkippedPerDatasource.forEach((dataSource, count) -> { |
Contributor
There was a problem hiding this comment.
Nit: Lambda can be simplified
| polledDataSourceMetadata.forEach(this::updateDSMetadata); | ||
|
|
||
| // Remove segments of the datasource from refresh list for which we received schema from the Coordinator. | ||
| final Map<String, Integer> segmentsSkippedPerDatasource = new HashMap<>(); |
Contributor
There was a problem hiding this comment.
Please put the entire new logic in a separate method.
kfaraz
reviewed
Sep 23, 2025
Comment on lines
+230
to
+239
| // Emit metrics per datasource | ||
| datasourceToNumSegmentsSkipped.forEach( | ||
| (dataSource, count) -> | ||
| emitMetric( | ||
| Metric.BROKER_SEGMENTS_SKIPPED_REFRESH, | ||
| count, | ||
| new ServiceMetricEvent.Builder().setDimension( | ||
| DruidMetrics.DATASOURCE, | ||
| dataSource)) | ||
| ); |
Contributor
There was a problem hiding this comment.
This should be put in the new method as well.
| } | ||
| } | ||
|
|
||
| private Map<String, Integer> getDatasourceToSegmentsSkipped( |
Contributor
There was a problem hiding this comment.
Suggested change
| private Map<String, Integer> getDatasourceToSegmentsSkipped( | |
| private void emitMetricForSkippedSegments( |
| datasourceToNumSegmentsSkipped.merge(segmentId.getDataSource(), 1, Integer::sum); | ||
| } | ||
| } | ||
| return datasourceToNumSegmentsSkipped; |
Contributor
There was a problem hiding this comment.
Don't return anything. Perform the metric emission here itself.
Comment on lines
+236
to
+238
| new ServiceMetricEvent.Builder().setDimension( | ||
| DruidMetrics.DATASOURCE, | ||
| dataSource)) |
Contributor
There was a problem hiding this comment.
Suggested change
| new ServiceMetricEvent.Builder().setDimension( | |
| DruidMetrics.DATASOURCE, | |
| dataSource)) | |
| new ServiceMetricEvent.Builder().setDimension(DruidMetrics.DATASOURCE, dataSource) | |
| ) |
Contributor
Author
|
Thanks for your review @kfaraz , I've accommodated the review comments, PTAL |
kfaraz
approved these changes
Sep 23, 2025
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
As follow-up of #18497 , this PR re-enables the embedded tests in
CentralizedSchemaMetadataQueryDisabledTestWeirdly, this is not unblocking
CentralizedSchemaPublishFailureTest(the no. of segments skipped and the no. of segments refreshed seems to be changing erratically in tests and is never what we want, will take a look at those tests in separate PR.)