kfaraz commented on code in PR #14507:
URL: https://github.com/apache/druid/pull/14507#discussion_r1251011107
##########
indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java:
##########
@@ -313,6 +326,10 @@ private List<TimelineObjectHolder<String, DataSegment>>
createTimeline()
if (interval == null) {
return getTimelineForSegmentIds(coordinatorClient, dataSource,
segmentIds);
} else {
+ if (toolbox != null) {
+ return getLockedTimelineForInterval(retryPolicyFactory, dataSource,
interval);
+ }
+
return getTimelineForInterval(coordinatorClient, retryPolicyFactory,
dataSource, interval);
}
Review Comment:
for readability:
```suggestion
} else if (toolbox == null) {
return getTimelineForInterval(coordinatorClient, retryPolicyFactory,
dataSource, interval);
} else {
return getLockedTimelineForInterval(retryPolicyFactory, dataSource,
interval);
}
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java:
##########
@@ -484,6 +501,54 @@ private static SortedMap<WindowedSegmentId, Long>
createWindowedSegmentIdFromTim
return segmentSizeMap;
}
+
+ public List<TimelineObjectHolder<String, DataSegment>>
getLockedTimelineForInterval(
+ RetryPolicyFactory retryPolicyFactory,
+ String dataSource,
+ Interval interval
+ )
+ {
+ Preconditions.checkNotNull(interval);
+
+ // This call used to use the TaskActionClient, so for compatibility we use
the same retry configuration
+ // as TaskActionClient.
+ final RetryPolicy retryPolicy = retryPolicyFactory.makeRetryPolicy();
+ Collection<DataSegment> usedSegments;
+ while (true) {
+ try {
+ usedSegments = toolbox.getTaskActionClient()
Review Comment:
If you need to use only the `TaskActionClient`, you could just do a
`@JacksonInject` of `TaskActionClientFactory` or `TaskMaster` in this class
instead of setting `TaskToolbox` everywhere.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsAction.java:
##########
@@ -58,17 +62,22 @@ public class RetrieveUsedSegmentsAction implements
TaskAction<Collection<DataSeg
@JsonIgnore
private final Segments visibility;
+ @JsonIgnore
+ private final boolean onlyLocked;
+
@JsonCreator
public RetrieveUsedSegmentsAction(
@JsonProperty("dataSource") String dataSource,
@Deprecated @JsonProperty("interval") Interval interval,
@JsonProperty("intervals") Collection<Interval> intervals,
// When JSON object is deserialized, this parameter is optional for
backward compatibility.
// Otherwise, it shouldn't be considered optional.
- @JsonProperty("visibility") @Nullable Segments visibility
+ @JsonProperty("visibility") @Nullable Segments visibility,
+ @JsonProperty("onlyLocked") @Nullable Boolean onlyLocked
Review Comment:
Rather than adding a new boolean property, we should try to incorporate this
into the definition of the `Segments` enum.
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/actions/RetrieveUsedSegmentsAction.java:
##########
@@ -114,8 +127,22 @@ public TypeReference<Collection<DataSegment>>
getReturnTypeReference()
@Override
public Collection<DataSegment> perform(Task task, TaskActionToolbox toolbox)
{
+ final Map<Interval, String> intervalToLockVersionMap = new HashMap<>();
Review Comment:
Please add comments here to explain the logic.
##########
indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java:
##########
@@ -313,6 +326,10 @@ private List<TimelineObjectHolder<String, DataSegment>>
createTimeline()
if (interval == null) {
return getTimelineForSegmentIds(coordinatorClient, dataSource,
segmentIds);
} else {
+ if (toolbox != null) {
Review Comment:
In which cases will this be `null` here?
##########
indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java:
##########
@@ -484,6 +501,54 @@ private static SortedMap<WindowedSegmentId, Long>
createWindowedSegmentIdFromTim
return segmentSizeMap;
}
+
+ public List<TimelineObjectHolder<String, DataSegment>>
getLockedTimelineForInterval(
+ RetryPolicyFactory retryPolicyFactory,
+ String dataSource,
+ Interval interval
+ )
+ {
+ Preconditions.checkNotNull(interval);
+
+ // This call used to use the TaskActionClient, so for compatibility we use
the same retry configuration
Review Comment:
This copied comment is invalid here as you _are_ using the
`TaskActionClient` in this case.
##########
indexing-service/src/main/java/org/apache/druid/indexing/input/DruidInputSource.java:
##########
@@ -484,6 +501,54 @@ private static SortedMap<WindowedSegmentId, Long>
createWindowedSegmentIdFromTim
return segmentSizeMap;
}
+
+ public List<TimelineObjectHolder<String, DataSegment>>
getLockedTimelineForInterval(
+ RetryPolicyFactory retryPolicyFactory,
+ String dataSource,
+ Interval interval
+ )
+ {
+ Preconditions.checkNotNull(interval);
+
+ // This call used to use the TaskActionClient, so for compatibility we use
the same retry configuration
+ // as TaskActionClient.
+ final RetryPolicy retryPolicy = retryPolicyFactory.makeRetryPolicy();
+ Collection<DataSegment> usedSegments;
+ while (true) {
+ try {
+ usedSegments = toolbox.getTaskActionClient()
Review Comment:
Going by the comment above, this code might have used `TaskActionClient`
earlier, you could also check how it was originally being done.
--
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]
