kfaraz commented on code in PR #19532:
URL: https://github.com/apache/druid/pull/19532#discussion_r3579171625
##########
server/src/main/java/org/apache/druid/metadata/SqlSegmentsMetadataQuery.java:
##########
@@ -1093,6 +1093,7 @@ public List<Interval>
retrieveUnusedSegmentIntervals(String dataSource, int limi
return
intervals.stream().filter(Objects::nonNull).collect(Collectors.toList());
}
+
Review Comment:
nit: not required
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/BalanceSegments.java:
##########
@@ -50,25 +51,49 @@ public BalanceSegments(Duration coordinatorPeriod)
@Override
public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams
params)
{
+ final Stopwatch totalTime = Stopwatch.createStarted();
+
if (params.getUsedSegmentCount() <= 0) {
+ log.info("BalanceSegments skipped: usedSegmentCount[%,d].",
params.getUsedSegmentCount());
Review Comment:
As already called out in the description, please ensure that all the logs
are removed before merging. Otherwise, coordinator logs will become very noisy.
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -64,6 +63,10 @@ public class StrategicSegmentAssigner implements
SegmentActionHandler
private final RoundRobinServerSelector serverSelector;
private final BalancerStrategy strategy;
+ // Fixed for the lifetime of this assigner (a new instance is created each
coordinator run),
+ // so it is computed once here rather than on every
replicateSegment/replicateSegmentPartially call.
Review Comment:
This comment can be omitted since we are already maintaining other similar
fields which are valid for the current run such as `historicalTierAliases` and
`tierToAliasName`.
##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java:
##########
@@ -782,11 +783,32 @@ private class UpdateReplicationStatus implements
CoordinatorDuty
@Override
public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams
params)
{
+ final Stopwatch totalTime = Stopwatch.createStarted();
+
+ final Stopwatch broadcastTime = Stopwatch.createStarted();
Review Comment:
Please remove these stopwatches as well before merging this PR.
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/StrategicSegmentAssigner.java:
##########
@@ -86,6 +89,9 @@ public StrategicSegmentAssigner(
this.cluster = cluster;
this.strategy = strategy;
this.loadQueueManager = loadQueueManager;
+ final Set<String> allTiersInCluster = new HashSet<>();
+ cluster.getTierNames().forEach(allTiersInCluster::add);
+ this.allTiersInCluster = allTiersInCluster;
Review Comment:
to ensure immutability:
```suggestion
this.allTiersInCluster = Set.copyOf(cluster.getTierNames());
```
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkOvershadowedSegmentsAsUnused.java:
##########
@@ -80,23 +80,33 @@ public DruidCoordinatorRuntimeParams
run(DruidCoordinatorRuntimeParams params)
return params;
}
+ // Identify the datasources that actually have overshadowed segments to
check.
+ // Timelines only need to be built for these datasources.
+ final Set<String> relevantDatasources = new HashSet<>();
+ for (DataSegment s : allOvershadowedSegments) {
+ relevantDatasources.add(s.getDataSource());
+ }
Review Comment:
Nit: rename and make immutable
```suggestion
final Set<String> eligibleDatasources = allOvershadowedSegments
.stream()
.map(DataSegment::getDataSource)
.collect(Collectors.toSet());
```
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkOvershadowedSegmentsAsUnused.java:
##########
@@ -80,23 +80,33 @@ public DruidCoordinatorRuntimeParams
run(DruidCoordinatorRuntimeParams params)
return params;
}
+ // Identify the datasources that actually have overshadowed segments to
check.
+ // Timelines only need to be built for these datasources.
+ final Set<String> relevantDatasources = new HashSet<>();
Review Comment:
```suggestion
final Set<String> eligibleDatasources = new HashSet<>();
```
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java:
##########
@@ -132,43 +130,57 @@ public DruidCoordinatorRuntimeParams run(final
DruidCoordinatorRuntimeParams par
private Map<String, Set<SegmentId>>
determineNonOvershadowedEternityTombstones(final DataSourcesSnapshot
dataSourcesSnapshot)
{
final Map<String, Set<SegmentId>>
datasourceToNonOvershadowedEternityTombstones = new HashMap<>();
+ final Map<String, Set<DataSegment>> overshadowedSegmentsByDatasource = new
HashMap<>();
- dataSourcesSnapshot.getDataSourcesMap().keySet().forEach((datasource) -> {
+ for (final DataSegment overshadowedSegment :
dataSourcesSnapshot.getOvershadowedSegments()) {
+ overshadowedSegmentsByDatasource
+ .computeIfAbsent(overshadowedSegment.getDataSource(), ds -> new
HashSet<>())
+ .add(overshadowedSegment);
+ }
+
+ for (final ImmutableDruidDataSource dataSource :
dataSourcesSnapshot.getDataSourcesWithAllUsedSegments()) {
+ final String datasource = dataSource.getName();
Review Comment:
To avoid confusion with the other variable `dataSource`.
```suggestion
final String datasourceName = dataSource.getName();
```
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicaCount.java:
##########
@@ -36,6 +36,25 @@ public class SegmentReplicaCount
private int movingTo;
private int movingFrom;
+ SegmentReplicaCount()
+ {
+ }
+
+ /**
+ * Copies the counts out of {@code other} into a new, independent instance.
+ */
+ SegmentReplicaCount(SegmentReplicaCount other)
Review Comment:
Instead of this, maybe add a `copy()` method.
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicaCount.java:
##########
@@ -36,6 +36,25 @@ public class SegmentReplicaCount
private int movingTo;
private int movingFrom;
+ SegmentReplicaCount()
Review Comment:
This shouldn't be needed.
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/PrepareBalancerAndLoadQueues.java:
##########
@@ -203,14 +203,14 @@ private void collectHistoricalStats(
private void collectUsedSegmentStats(DruidCoordinatorRuntimeParams params,
CoordinatorRunStats stats)
{
- params.getUsedSegmentsTimelinesPerDataSource().forEach((dataSource,
timeline) -> {
- long totalSizeOfUsedSegments = timeline.iterateAllObjects().stream()
-
.mapToLong(DataSegment::getSize).sum();
+ for (final ImmutableDruidDataSource dataSource :
params.getDataSourcesSnapshot().getDataSourcesWithAllUsedSegments()) {
+ final long totalSizeOfUsedSegments = dataSource.getTotalSizeOfSegments();
+ final int numUsedSegments = dataSource.getSegments().size();
Review Comment:
We should also add a method `DataSourcesSnapshot.getNumSegments()` to avoid
materializing the collection inside `DataSourcesSnapshot.getSegments()`.
##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java:
##########
@@ -801,26 +803,46 @@ public DruidCoordinatorRuntimeParams
run(DruidCoordinatorRuntimeParams params)
{
// Collect stats for unavailable and under-replicated segments
final CoordinatorRunStats stats = params.getCoordinatorStats();
- getDatasourceToUnavailableSegmentCount().forEach(
- (dataSource, numUnavailable) -> stats.add(
- Stats.Segments.UNAVAILABLE,
- RowKey.of(Dimension.DATASOURCE, dataSource),
- numUnavailable
- )
- );
- getTierToDatasourceToUnderReplicatedCount(false).forEach(
- (tier, countsPerDatasource) -> countsPerDatasource.forEach(
- (dataSource, underReplicatedCount) ->
- stats.addToSegmentStat(Stats.Segments.UNDER_REPLICATED,
tier, dataSource, underReplicatedCount)
- )
- );
- getDatasourceToDeepStorageQueryOnlySegmentCount().forEach(
- (dataSource, numDeepStorageOnly) -> stats.add(
- Stats.Segments.DEEP_STORAGE_ONLY,
- RowKey.of(Dimension.DATASOURCE, dataSource),
- numDeepStorageOnly
- )
- );
+
+ final Object2IntMap<String> dsToUnavailable;
+ final Map<String, Object2LongMap<String>> tierToDsToUnderRepl;
+ final Object2IntMap<String> dsToDeepStorageOnly;
+
+ if (segmentReplicationStatus == null) {
+ dsToUnavailable = Object2IntMaps.emptyMap();
+ tierToDsToUnderRepl = Collections.emptyMap();
+ dsToDeepStorageOnly = Object2IntMaps.emptyMap();
+ } else {
+ // Single fused pass replaces three independent full iterations over
+ // metadataManager.iterateAllUsedSegments(). ignoreMissingServers=true
replicates
+ // the !false inversion inside computeUnderReplicated(segments, false).
+ final SegmentReplicationStatus.SegmentStatsSnapshot snapshot =
segmentReplicationStatus.computeSegmentStats(
+ metadataManager.iterateAllUsedSegments(),
+ true
+ );
+ dsToUnavailable = snapshot.getDatasourceToUnavailableCount();
+ tierToDsToUnderRepl =
snapshot.getTierToDatasourceToUnderReplicatedCount();
+ dsToDeepStorageOnly = snapshot.getDatasourceToDeepStorageOnlyCount();
+ }
+
+ for (final Object2IntMap.Entry<String> e :
dsToUnavailable.object2IntEntrySet()) {
+ stats.add(Stats.Segments.UNAVAILABLE, RowKey.of(Dimension.DATASOURCE,
e.getKey()), e.getIntValue());
+ }
+
+ for (final Map.Entry<String, Object2LongMap<String>> tierEntry :
tierToDsToUnderRepl.entrySet()) {
Review Comment:
Using `forEach` and `fastForEach` instead of traditional for loops may make
this more readable.
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java:
##########
@@ -132,43 +130,57 @@ public DruidCoordinatorRuntimeParams run(final
DruidCoordinatorRuntimeParams par
private Map<String, Set<SegmentId>>
determineNonOvershadowedEternityTombstones(final DataSourcesSnapshot
dataSourcesSnapshot)
{
final Map<String, Set<SegmentId>>
datasourceToNonOvershadowedEternityTombstones = new HashMap<>();
+ final Map<String, Set<DataSegment>> overshadowedSegmentsByDatasource = new
HashMap<>();
- dataSourcesSnapshot.getDataSourcesMap().keySet().forEach((datasource) -> {
+ for (final DataSegment overshadowedSegment :
dataSourcesSnapshot.getOvershadowedSegments()) {
+ overshadowedSegmentsByDatasource
+ .computeIfAbsent(overshadowedSegment.getDataSource(), ds -> new
HashSet<>())
+ .add(overshadowedSegment);
+ }
+
+ for (final ImmutableDruidDataSource dataSource :
dataSourcesSnapshot.getDataSourcesWithAllUsedSegments()) {
+ final String datasource = dataSource.getName();
final SegmentTimeline usedSegmentsTimeline
=
dataSourcesSnapshot.getUsedSegmentsTimelinesPerDataSource().get(datasource);
+ if (usedSegmentsTimeline == null) {
Review Comment:
Can this ever happen? We are already iterating over the datasources provided
by the snapshot itself.
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/MarkEternityTombstonesAsUnused.java:
##########
@@ -132,43 +130,57 @@ public DruidCoordinatorRuntimeParams run(final
DruidCoordinatorRuntimeParams par
private Map<String, Set<SegmentId>>
determineNonOvershadowedEternityTombstones(final DataSourcesSnapshot
dataSourcesSnapshot)
{
final Map<String, Set<SegmentId>>
datasourceToNonOvershadowedEternityTombstones = new HashMap<>();
+ final Map<String, Set<DataSegment>> overshadowedSegmentsByDatasource = new
HashMap<>();
- dataSourcesSnapshot.getDataSourcesMap().keySet().forEach((datasource) -> {
+ for (final DataSegment overshadowedSegment :
dataSourcesSnapshot.getOvershadowedSegments()) {
+ overshadowedSegmentsByDatasource
+ .computeIfAbsent(overshadowedSegment.getDataSource(), ds -> new
HashSet<>())
+ .add(overshadowedSegment);
+ }
+
+ for (final ImmutableDruidDataSource dataSource :
dataSourcesSnapshot.getDataSourcesWithAllUsedSegments()) {
+ final String datasource = dataSource.getName();
final SegmentTimeline usedSegmentsTimeline
=
dataSourcesSnapshot.getUsedSegmentsTimelinesPerDataSource().get(datasource);
+ if (usedSegmentsTimeline == null) {
+ continue;
+ }
- final Optional<Set<DataSegment>> usedNonOvershadowedSegments =
- Optional.fromNullable(usedSegmentsTimeline)
- .transform(timeline ->
timeline.findNonOvershadowedObjectsInInterval(
- Intervals.ETERNITY,
- Partitions.ONLY_COMPLETE
- ));
-
- if (usedNonOvershadowedSegments.isPresent()) {
- usedNonOvershadowedSegments.get().forEach(candidateSegment -> {
- if (isNewGenerationEternityTombstone(candidateSegment)) {
- boolean overlaps =
dataSourcesSnapshot.getOvershadowedSegments().stream()
- .filter(overshadowedSegment
->
-
candidateSegment.getDataSource()
-
.equals(overshadowedSegment.getDataSource()))
- .anyMatch(
- overshadowedSegment ->
-
candidateSegment.getInterval()
-
.overlaps(overshadowedSegment.getInterval())
- );
- if (!overlaps) {
- datasourceToNonOvershadowedEternityTombstones
- .computeIfAbsent(datasource, ds -> new HashSet<>())
- .add(candidateSegment.getId());
- }
- }
- });
+ for (final DataSegment candidateSegment : dataSource.getSegments()) {
+ if (isNewGenerationEternityTombstone(candidateSegment)
+ && !usedSegmentsTimeline.isOvershadowed(candidateSegment)
+ && !overlapsAnyOvershadowedSegment(
+ candidateSegment,
+
overshadowedSegmentsByDatasource.get(candidateSegment.getDataSource())
+ )) {
Review Comment:
Let's collapse these 3 conditions into a single new method for readability
of this for loop:
```
shouldMarkAsUnused(candidateSegment, overshadowedSegmentsByDatasource,
timeline)
```
the new method can check the 3 required conditions by invoking respective
methods.
##########
server/src/main/java/org/apache/druid/server/coordinator/DruidCoordinator.java:
##########
@@ -801,26 +803,46 @@ public DruidCoordinatorRuntimeParams
run(DruidCoordinatorRuntimeParams params)
{
// Collect stats for unavailable and under-replicated segments
final CoordinatorRunStats stats = params.getCoordinatorStats();
- getDatasourceToUnavailableSegmentCount().forEach(
- (dataSource, numUnavailable) -> stats.add(
- Stats.Segments.UNAVAILABLE,
- RowKey.of(Dimension.DATASOURCE, dataSource),
- numUnavailable
- )
- );
- getTierToDatasourceToUnderReplicatedCount(false).forEach(
- (tier, countsPerDatasource) -> countsPerDatasource.forEach(
- (dataSource, underReplicatedCount) ->
- stats.addToSegmentStat(Stats.Segments.UNDER_REPLICATED,
tier, dataSource, underReplicatedCount)
- )
- );
- getDatasourceToDeepStorageQueryOnlySegmentCount().forEach(
- (dataSource, numDeepStorageOnly) -> stats.add(
- Stats.Segments.DEEP_STORAGE_ONLY,
- RowKey.of(Dimension.DATASOURCE, dataSource),
- numDeepStorageOnly
- )
- );
+
+ final Object2IntMap<String> dsToUnavailable;
+ final Map<String, Object2LongMap<String>> tierToDsToUnderRepl;
+ final Object2IntMap<String> dsToDeepStorageOnly;
Review Comment:
Please use full names e.g. `datasourceToUnavailableCount`,
`datasourceToUnderReplicatedCount`.
Do we need to keep 3 separate map references? Maybe just use an empty
instance of `SegmentStatsSnapshot` and invoke getters on it where needed.
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -40,15 +43,27 @@ public class SegmentReplicationStatus
public SegmentReplicationStatus(Map<SegmentId, Map<String,
SegmentReplicaCount>> replicaCountsInTier)
{
- this.replicaCountsInTier = ImmutableMap.copyOf(replicaCountsInTier);
-
- final Map<SegmentId, SegmentReplicaCount> totalReplicaCounts = new
HashMap<>();
- replicaCountsInTier.forEach((segmentId, tierToReplicaCount) -> {
+ // replicaCountsInTier is the caller's live SegmentReplicaCountMap and is
mutated further
+ // in the same coordinator cycle (e.g. by BalanceSegments), so we must
snapshot both the
+ // structure and the per-tier SegmentReplicaCount values, not just alias
the outer map.
Review Comment:
Nit: simpler language
```suggestion
// Make a deep copy of replicaCountsInTier as it may be mutated further
// in the same coordinator cycle (e.g. by BalanceSegments)
```
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/RunRules.java:
##########
@@ -98,14 +99,15 @@ public DruidCoordinatorRuntimeParams
run(DruidCoordinatorRuntimeParams params)
final DateTime now = DateTimes.nowUtc();
final Object2IntOpenHashMap<String> datasourceToSegmentsWithNoRule = new
Object2IntOpenHashMap<>();
+ final Map<String, List<Rule>> datasourceToRules = new HashMap<>();
Review Comment:
Rules need not be cached as they are already served from the cache in
`SQLMetadataRuleManager` itself. The changes to this class may be reverted.
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -81,4 +96,84 @@ public Map<String, Object2LongMap<String>>
getTierToDatasourceToUnderReplicated(
return tierToUnderReplicated;
}
+
+ /**
+ * Computes unavailable, under-replicated and deep-storage-only segment
counts in a single
+ * pass over {@code usedSegments}, instead of three independent full
iterations. Produces
+ * results identical to calling {@link #getReplicaCountsInCluster} and
+ * {@link #getTierToDatasourceToUnderReplicated} independently for each
segment.
+ *
+ * @param ignoreMissingServers same semantics as in {@link
#getTierToDatasourceToUnderReplicated}.
+ */
+ public SegmentStatsSnapshot computeSegmentStats(Iterable<DataSegment>
usedSegments, boolean ignoreMissingServers)
+ {
+ final Object2IntOpenHashMap<String> datasourceToUnavailable = new
Object2IntOpenHashMap<>();
+ final Object2IntOpenHashMap<String> datasourceToDeepStorageOnly = new
Object2IntOpenHashMap<>();
+ final Map<String, Object2LongMap<String>> tierToUnderReplicated = new
HashMap<>();
+
+ for (DataSegment segment : usedSegments) {
+ final SegmentId segmentId = segment.getId();
+ final String datasource = segment.getDataSource();
+
+ final SegmentReplicaCount totalCount = totalReplicaCounts.get(segmentId);
+ if (totalCount != null && (totalCount.totalLoaded() > 0 ||
totalCount.required() == 0)) {
+ datasourceToUnavailable.addTo(datasource, 0);
+ } else {
+ datasourceToUnavailable.addTo(datasource, 1);
+ }
+ if (totalCount != null && totalCount.totalLoaded() == 0 &&
totalCount.required() == 0) {
+ datasourceToDeepStorageOnly.addTo(datasource, 1);
+ }
+
+ final Map<String, SegmentReplicaCount> tierToReplicaCount =
replicaCountsInTier.get(segmentId);
+ if (tierToReplicaCount != null) {
+ tierToReplicaCount.forEach((tier, counts) -> {
+ final int underReplicated = ignoreMissingServers ? counts.missing()
: counts.missingAndLoadable();
+ if (underReplicated >= 0) {
+ Object2LongOpenHashMap<String> datasourceToUnderReplicated =
(Object2LongOpenHashMap<String>)
+ tierToUnderReplicated.computeIfAbsent(tier, ds -> new
Object2LongOpenHashMap<>());
+ datasourceToUnderReplicated.addTo(datasource, underReplicated);
+ }
+ });
Review Comment:
Can we replace this by a call to `getTierToDatasourceToUnderReplicated()`
since it is the same logic?
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -81,4 +96,84 @@ public Map<String, Object2LongMap<String>>
getTierToDatasourceToUnderReplicated(
return tierToUnderReplicated;
}
+
+ /**
+ * Computes unavailable, under-replicated and deep-storage-only segment
counts in a single
+ * pass over {@code usedSegments}, instead of three independent full
iterations. Produces
+ * results identical to calling {@link #getReplicaCountsInCluster} and
+ * {@link #getTierToDatasourceToUnderReplicated} independently for each
segment.
+ *
+ * @param ignoreMissingServers same semantics as in {@link
#getTierToDatasourceToUnderReplicated}.
+ */
+ public SegmentStatsSnapshot computeSegmentStats(Iterable<DataSegment>
usedSegments, boolean ignoreMissingServers)
+ {
+ final Object2IntOpenHashMap<String> datasourceToUnavailable = new
Object2IntOpenHashMap<>();
+ final Object2IntOpenHashMap<String> datasourceToDeepStorageOnly = new
Object2IntOpenHashMap<>();
+ final Map<String, Object2LongMap<String>> tierToUnderReplicated = new
HashMap<>();
+
+ for (DataSegment segment : usedSegments) {
+ final SegmentId segmentId = segment.getId();
+ final String datasource = segment.getDataSource();
+
+ final SegmentReplicaCount totalCount = totalReplicaCounts.get(segmentId);
+ if (totalCount != null && (totalCount.totalLoaded() > 0 ||
totalCount.required() == 0)) {
+ datasourceToUnavailable.addTo(datasource, 0);
+ } else {
+ datasourceToUnavailable.addTo(datasource, 1);
+ }
+ if (totalCount != null && totalCount.totalLoaded() == 0 &&
totalCount.required() == 0) {
+ datasourceToDeepStorageOnly.addTo(datasource, 1);
+ }
+
+ final Map<String, SegmentReplicaCount> tierToReplicaCount =
replicaCountsInTier.get(segmentId);
+ if (tierToReplicaCount != null) {
+ tierToReplicaCount.forEach((tier, counts) -> {
+ final int underReplicated = ignoreMissingServers ? counts.missing()
: counts.missingAndLoadable();
+ if (underReplicated >= 0) {
+ Object2LongOpenHashMap<String> datasourceToUnderReplicated =
(Object2LongOpenHashMap<String>)
+ tierToUnderReplicated.computeIfAbsent(tier, ds -> new
Object2LongOpenHashMap<>());
+ datasourceToUnderReplicated.addTo(datasource, underReplicated);
+ }
+ });
+ }
+ }
+
+ return new SegmentStatsSnapshot(datasourceToUnavailable,
tierToUnderReplicated, datasourceToDeepStorageOnly);
+ }
+
+ /**
+ * Holder for the three segment-stat views computed together by {@link
#computeSegmentStats}.
+ */
+ public static class SegmentStatsSnapshot
Review Comment:
Please move out this class into a separate file as it seems completely
independent.
##########
server/src/main/java/org/apache/druid/server/coordinator/loading/SegmentReplicationStatus.java:
##########
@@ -81,4 +96,84 @@ public Map<String, Object2LongMap<String>>
getTierToDatasourceToUnderReplicated(
return tierToUnderReplicated;
}
+
+ /**
+ * Computes unavailable, under-replicated and deep-storage-only segment
counts in a single
+ * pass over {@code usedSegments}, instead of three independent full
iterations. Produces
+ * results identical to calling {@link #getReplicaCountsInCluster} and
+ * {@link #getTierToDatasourceToUnderReplicated} independently for each
segment.
+ *
+ * @param ignoreMissingServers same semantics as in {@link
#getTierToDatasourceToUnderReplicated}.
+ */
+ public SegmentStatsSnapshot computeSegmentStats(Iterable<DataSegment>
usedSegments, boolean ignoreMissingServers)
+ {
+ final Object2IntOpenHashMap<String> datasourceToUnavailable = new
Object2IntOpenHashMap<>();
+ final Object2IntOpenHashMap<String> datasourceToDeepStorageOnly = new
Object2IntOpenHashMap<>();
+ final Map<String, Object2LongMap<String>> tierToUnderReplicated = new
HashMap<>();
+
+ for (DataSegment segment : usedSegments) {
+ final SegmentId segmentId = segment.getId();
+ final String datasource = segment.getDataSource();
+
+ final SegmentReplicaCount totalCount = totalReplicaCounts.get(segmentId);
+ if (totalCount != null && (totalCount.totalLoaded() > 0 ||
totalCount.required() == 0)) {
+ datasourceToUnavailable.addTo(datasource, 0);
+ } else {
+ datasourceToUnavailable.addTo(datasource, 1);
+ }
+ if (totalCount != null && totalCount.totalLoaded() == 0 &&
totalCount.required() == 0) {
+ datasourceToDeepStorageOnly.addTo(datasource, 1);
+ }
Review Comment:
This logic is now duplicated with `DruidCoordinator`.
Please move unavailable logic and deep storage logic into new methods in
this class so that `DruidCoordinator` may reuse it too.
```java
boolean isUnavailable(SegmentId segmentId);
boolean isUnderReplicated(SegmentId segmentId);
```
--
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]
