pjain1 commented on a change in pull request #10524:
URL: https://github.com/apache/druid/pull/10524#discussion_r580207110
##########
File path:
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java
##########
@@ -318,6 +324,114 @@ public void handle()
}
}
+ // change taskCount without resubmitting.
+ private class DynamicAllocationTasksNotice implements Notice
+ {
+ Callable<Integer> scaleAction;
+
+ DynamicAllocationTasksNotice(Callable<Integer> scaleAction)
+ {
+ this.scaleAction = scaleAction;
+ }
+
+ /**
+ * This method will do lags points collection and check dynamic scale
action is necessary or not.
+ */
+ @Override
+ public void handle()
+ {
+ try {
+ long nowTime = System.currentTimeMillis();
+ // Only queue is full and over minTriggerDynamicFrequency can trigger
scale out/in
+ if (spec.isSuspended()) {
+ log.info("[%s] supervisor is suspended, skip to check dynamic
allocate task logic", dataSource);
+ return;
+ }
+ log.debug("PendingCompletionTaskGroups is [%s] for dataSource [%s].",
pendingCompletionTaskGroups, dataSource);
+ for (CopyOnWriteArrayList list : pendingCompletionTaskGroups.values())
{
+ if (!list.isEmpty()) {
+ log.info("Still hand off tasks unfinished, skip to do scale action
[%s] for dataSource [%s].", pendingCompletionTaskGroups, dataSource);
+ return;
+ }
+ }
+ if (autoScalerConfig != null && nowTime - dynamicTriggerLastRunTime <
autoScalerConfig.getMinTriggerDynamicFrequencyMillis()) {
+ log.info("NowTime - dynamicTriggerLastRunTime is [%s]. Defined
minTriggerDynamicFrequency is [%s] for dataSource [%s], CALM DOWN NOW !",
nowTime - dynamicTriggerLastRunTime,
autoScalerConfig.getMinTriggerDynamicFrequencyMillis(), dataSource);
+ return;
+ }
+
+ Integer desriedTaskCount = scaleAction.call();
+ boolean allocationSuccess = dynamicAllocate(desriedTaskCount);
+
+ if (allocationSuccess) {
+ dynamicTriggerLastRunTime = nowTime;
+ }
+ }
+ catch (Exception ex) {
+ log.warn(ex, "Error parsing DynamicAllocationTasksNotice");
+ }
+ }
+ }
+
+ /**
+ * This method determines how to do scale actions based on collected lag
points.
+ * If scale action is triggered :
+ * First of all, call gracefulShutdownInternal() which will change the
state of current datasource ingest tasks from reading to publishing.
+ * Secondly, clear all the stateful data structures:
activelyReadingTaskGroups, partitionGroups, partitionOffsets,
pendingCompletionTaskGroups, partitionIds. These structures will be rebuiled
next 'RunNotice'.
+ * Finally, change taskCount in SeekableStreamSupervisorIOConfig and sync
it to MetaStorage.
+ * After changed taskCount in SeekableStreamSupervisorIOConfig, next
RunNotice will create scaled number of ingest tasks without resubmitting
supervisors.
+ * @param desiredActiveTaskCount desired taskCount compute from autoscaler
+ * @return Boolean flag, do scale action successfully or not. If true , it
will take at least 'minTriggerDynamicFrequency' before next 'dynamicAllocate'.
+ * If false, it will do 'dynamicAllocate' again after
'dynamicCheckPeriod'.
Review comment:
```suggestion
* First of all, call gracefulShutdownInternal() which will change the
state of current datasource ingest tasks from reading to publishing.
* Secondly, clear all the stateful data structures:
activelyReadingTaskGroups, partitionGroups, partitionOffsets,
pendingCompletionTaskGroups, partitionIds. These structures will be rebuiled in
the next 'RunNotice'.
* Finally, change the taskCount in SeekableStreamSupervisorIOConfig
and sync it to MetadataStorage.
* After the taskCount si changed in SeekableStreamSupervisorIOConfig,
next RunNotice will create scaled number of ingest tasks without resubmitting
the supervisor.
* @param desiredActiveTaskCount desired taskCount compute from autoscaler
* @return Boolean flag indicating if scale action was executed or not. If
true, it will wait at least 'minTriggerDynamicFrequency' before next
'dynamicAllocate'.
* If false, it will do 'dynamicAllocate' again after
'dynamicCheckPeriod' millis.
```
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
