Restrict segment metadata kill query till maxInterval from last kill task time

server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnusedSegments.java

@@ -273,9 +273,21 @@ private Interval findIntervalForKill(
   )
   {
     final DateTime minStartTime = datasourceToLastKillIntervalEnd.get(dataSource);
-    final DateTime maxEndTime = ignoreDurationToRetain
-                                ? DateTimes.COMPARE_DATE_AS_STRING_MAX
-                                : DateTimes.nowUtc().minus(durationToRetain);
+
+    // Once the first segment from a datasource is killed, we have a valid minStartTime.
+    // Restricting the upper bound to scan segments metadata while running the kill task results in a efficient SQL query.
+    final DateTime maxEndTime;
+    if (ignoreDurationToRetain) {
+      maxEndTime = DateTimes.COMPARE_DATE_AS_STRING_MAX;
+    } else if (minStartTime == null) {
+      maxEndTime = DateTimes.nowUtc().minus(durationToRetain);
+    } else {
+      // If we have already killed a segment, limit the kill interval based on the minStartTime
+      maxEndTime = DateTimes.min(
+              DateTimes.nowUtc().minus(durationToRetain),
+              minStartTime.plus(maxIntervalToKill)
+      );
+    }
 
     final List<Interval> unusedSegmentIntervals = limitToPeriod(
         segmentsMetadataManager.getUnusedSegmentIntervals(

server/src/test/java/org/apache/druid/server/coordinator/duty/KillUnusedSegmentsTest.java

@@ -78,6 +78,7 @@ public class KillUnusedSegmentsTest
   private static final DateTime NOW = DateTimes.nowUtc();
   private static final Interval YEAR_OLD = new Interval(Period.days(1), NOW.minusDays(365));
   private static final Interval MONTH_OLD = new Interval(Period.days(1), NOW.minusDays(30));
+  private static final Interval FIFTEEN_DAY_OLD = new Interval(Period.days(1), NOW.minusDays(15));
   private static final Interval DAY_OLD = new Interval(Period.days(1), NOW.minusDays(1));
   private static final Interval HOUR_OLD = new Interval(Period.days(1), NOW.minusHours(1));
   private static final Interval NEXT_DAY = new Interval(Period.days(1), NOW.plusDays(1));
@@ -604,6 +605,55 @@ public void testLowerMaxIntervalToKill()
     validateLastKillStateAndReset(DS1, YEAR_OLD);
   }
 
+  @Test
+  public void testMaxIntervalToKillOverridesDurationToRetain()
+  {
+    configBuilder.withDurationToRetain(Period.hours(6).toStandardDuration())
+            .withMaxIntervalToKill(Period.days(20));
+
+    initDuty();
+
+    createAndAddUnusedSegment(DS1, MONTH_OLD, VERSION, NOW.minusDays(29));
+    CoordinatorRunStats newDatasourceStats = runDutyAndGetStats();
+
+    // For a new datasource, the duration to retain is used to determine kill interval
+    Assert.assertEquals(1, newDatasourceStats.get(Stats.Kill.ELIGIBLE_UNUSED_SEGMENTS, DS1_STAT_KEY));
+    validateLastKillStateAndReset(DS1, MONTH_OLD);
+
+    // For a datasource where kill has already happened, maxIntervalToKill is used
+    // if it leads to a smaller kill interval than durationToRetain
+    createAndAddUnusedSegment(DS1, FIFTEEN_DAY_OLD, VERSION, NOW.minusDays(14));
+    createAndAddUnusedSegment(DS1, DAY_OLD, VERSION, NOW.minusHours(2));
+    CoordinatorRunStats oldDatasourceStats = runDutyAndGetStats();
+
+    Assert.assertEquals(2, oldDatasourceStats.get(Stats.Kill.ELIGIBLE_UNUSED_SEGMENTS, DS1_STAT_KEY));
+    validateLastKillStateAndReset(DS1, FIFTEEN_DAY_OLD);
+  }
+
+  @Test
+  public void testDurationToRetainOverridesMaxIntervalToKill()
+  {
+    configBuilder.withDurationToRetain(Period.days(20).toStandardDuration())
+            .withMaxIntervalToKill(Period.days(350));
+
+    initDuty();
+
+    createAndAddUnusedSegment(DS1, YEAR_OLD, VERSION, NOW.minusDays(29));
+    CoordinatorRunStats newDatasourceStats = runDutyAndGetStats();
+
+    Assert.assertEquals(1, newDatasourceStats.get(Stats.Kill.ELIGIBLE_UNUSED_SEGMENTS, DS1_STAT_KEY));
+    validateLastKillStateAndReset(DS1, YEAR_OLD);
+
+    // For a datasource where (now - durationToRetain) < (lastKillTime(year old segment) + maxInterval)
+    // Fifteen day old segment will be rejected
+    createAndAddUnusedSegment(DS1, MONTH_OLD, VERSION, NOW.minusDays(29));
+    createAndAddUnusedSegment(DS1, FIFTEEN_DAY_OLD, VERSION, NOW.minusDays(14));
+    CoordinatorRunStats oldDatasourceStats = runDutyAndGetStats();
+
+    Assert.assertEquals(2, oldDatasourceStats.get(Stats.Kill.ELIGIBLE_UNUSED_SEGMENTS, DS1_STAT_KEY));
+    validateLastKillStateAndReset(DS1, MONTH_OLD);
+  }
+
   @Test
   public void testHigherMaxIntervalToKill()
   {