Add comment about Future to FixedSourcePartitionedScheduler
Add comment to non-obvious design choices made to avoid
ListenableFuture object retention problem.
@@ -167,10 +167,12 @@ public class FixedSourcePartitionedScheduler
BlockedReason blockedReason = BlockedReason.NO_ACTIVE_DRIVER_GROUP;
if (groupedLifespanScheduler.isPresent()) {
- // start new driver groups on the first scheduler, if previous ones have finished execution
- // (Note: finished execution is different from finished scheduling)
- SettableFuture newDriverGroupReady = groupedLifespanScheduler.get().schedule(sourcePartitionedSchedulers.get(0));
- blocked.add(newDriverGroupReady);
+ // Start new driver groups on the first scheduler if necessary,
+ // i.e. when previous ones have finished execution (not finished scheduling).
+ //
+ // Invoke schedule method to get a new SettableFuture every time.
+ // Reusing previously returned SettableFuture could lead to the ListenableFuture retaining too many listeners.
+ blocked.add(groupedLifespanScheduler.get().schedule(sourcePartitionedSchedulers.get(0)));
}
int splitsScheduled = 0;
@@ -319,6 +321,9 @@ public class FixedSourcePartitionedScheduler
public SettableFuture schedule(SourcePartitionedScheduler scheduler)
{
+ // Return a new future even if newDriverGroupReady has not finished.
+ // Returning the same SettableFuture instance could lead to ListenableFuture retaining too many listener objects.
+
checkState(initialScheduled);
List<Lifespan> recentlyCompletedDriverGroups;
