Consolidate RetrieveSegmentsToReplaceAction into RetrieveUsedSegmentsAction

[AmatyaAvadhanula]

This PR aims to let the original RetrieveUsedSegmentsAction work with REPLACE locks by introducing a boolean flag.
When the task does not hold any REPLACE locks, the behaviour is identical to the existing behaviour.
However, when there are REPLACE locks, only segments that were created before the times corresponding to the versions of the locks will be fetched for their respective intervals.

Using the original action helps with rolling upgrades where the flag is simply ignored when the overlord has not been upgraded.
It also helps eliminate the need for the undocumented parameter introduced in #15430

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[kfaraz]

We should also remove the other task action RetrieveSegmentsToReplaceAction as that is not needed anymore.
A rolling upgrade would upgrade MMs first, so we would not have an issue of firing a task action that the OL doesn't recognize.

[kfaraz]

Can this not be determined automatically while running the action? Only tasks with a REPLACE lock would have this as true, right?

[AmatyaAvadhanula]

Thanks, I think this can always be true.
If we need to fetch all the unfiltered segments for a task holding replace locks, we could use a task action client created using a different task that holds no locks.

[kfaraz]

Cleaner to do:

if (isReplaceTask()) {
  return newMethodWhichDoesTheRightThingForReplaceTask();
} else {
  return retrieveUsedSegments(toolbox);
}

private boolean isReplaceTask() {
   return replace && task.getDatasource().equals(dataSource);
}

[kfaraz]

The new method should also have a javadoc saying that it returns a consistent view of segments and why it is needed.

[AmatyaAvadhanula]

I think the current structure makes it more readable as it is obvious that we are using the old logic whenever possible before trying to filter the segments using the locks.

if (datasourceToReplace != datasourceToRead) {
  return retrieveUsedSegments();
}

Set<ReplaceLock> replaceLocks = fetchReplaceLocksForTask();

if (replaceLocks.isEmpty()) {
  return retrieveUsedSegments()
} 

return retrieveUsedSegmentsCreatedBeforeReplaceVersions(replaceLocks);

[kfaraz]

For later, can we confirm if this logic is really needed? I think the task action is always fired using the supervisor task ID.

[kfaraz]

Instead of logging all segment IDs separately, add them to a set and just log once.

[AmatyaAvadhanula]

The intent was to keep it verbose with the segment id and created date with the replace version available in each log.

[kfaraz]

I fear it might end up being too verbose. Better to just log the replace lock version for each interval and point out that anything newer than that would not be considered.
The segment IDs if needed can go in a debug log.

[kfaraz]

Changes look okay, left minor comments and MSQ tests need to be handled.

[kfaraz]

Don't think this comment is really needed or accurate. The real reason we are not firing a task action if the intervals is empty because we know we would get back an empty result. Why perform an unnecessary round trip?

[kfaraz]

style:

Suggested change

	publishedUsedSegments = context.taskActionClient().submit(new RetrieveUsedSegmentsAction(
	dataSource,
	intervals
	));
	publishedUsedSegments = context.taskActionClient().submit(
	new RetrieveUsedSegmentsAction(dataSource, intervals)
	);

[kfaraz]

I recall putting a comment here but can't find it anywhere.
I would prefer it if we moved the code here on down into a new method
retrieveUsedSegmentsForReplace(toolbox, replaceLocks).