zhangyue19921010 commented on a change in pull request #10524:
URL: https://github.com/apache/druid/pull/10524#discussion_r571047852
##########
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], CLAM 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, when parse 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 ceate scaled number of ingest tasks without resubmitting
supervisors.
+ * @param desireActiveTaskCount desire taskCount compute from autoscaler
+ * @return Boolean flag, do scale action successfully or not. If true , it
will take at least 'minTriggerDynamicFrequency' before next 'dynamicAllocatie'.
Review comment:
Changed.
##########
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], CLAM DOWN NOW !",
nowTime - dynamicTriggerLastRunTime,
autoScalerConfig.getMinTriggerDynamicFrequencyMillis(), dataSource);
Review comment:
Changed.
##########
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], CLAM 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, when parse DynamicAllocationTasksNotice");
Review comment:
Done.
##########
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], CLAM 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, when parse 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 ceate scaled number of ingest tasks without resubmitting
supervisors.
Review comment:
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.
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
