You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "Fly-Style (via GitHub)" <gi...@apache.org> on 2025/11/14 09:41:04 UTC
[PR] Improve lag-based autoscaler config persistence (druid)
Fly-Style opened a new pull request, #18745:
URL: https://github.com/apache/druid/pull/18745
- When the autoscaler does a scaling action, it updates `ioConfig.autoScalerConfig.taskCountStart`, not `ioConfig.taskCount`
- When a supervisor is updated via API, and `ioConfig.autoScalerConfig.taskCountStart` is not specified, `ioConfig.autoScalerConfig.taskCountStart` is merged from the currently-persisted supervisor.
##### Key changed/added classes in this PR
* `SupervisorManager`
* `SeekableStreamSupervisorSpec`
This PR has:
- [ ] been self-reviewed.
- [ ] 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](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [x] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] been tested in a test Druid cluster.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18745:
URL: https://github.com/apache/druid/pull/18745#discussion_r2610496869
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -556,43 +566,51 @@ public String getType()
private boolean changeTaskCount(int desiredActiveTaskCount)
throws InterruptedException, ExecutionException
{
+ if (autoScalerConfig == null) {
+ log.warn("autoScalerConfig is 'null' but dynamic allocation notice is submitted, how can it be ?");
+ return false;
+ }
int currentActiveTaskCount;
Collection<TaskGroup> activeTaskGroups = activelyReadingTaskGroups.values();
currentActiveTaskCount = activeTaskGroups.size();
if (desiredActiveTaskCount < 0 || desiredActiveTaskCount == currentActiveTaskCount) {
return false;
- } else {
- log.info(
- "Starting scale action, current active task count is [%d] and desired task count is [%d] for supervisor[%s] for dataSource[%s].",
- currentActiveTaskCount,
- desiredActiveTaskCount,
- supervisorId,
- dataSource
- );
- final Stopwatch scaleActionStopwatch = Stopwatch.createStarted();
- gracefulShutdownInternal();
- changeTaskCountInIOConfig(desiredActiveTaskCount);
- clearAllocationInfo();
- emitter.emit(ServiceMetricEvent.builder()
- .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
- .setDimension(DruidMetrics.DATASOURCE, dataSource)
- .setDimension(DruidMetrics.STREAM, getIoConfig().getStream())
- .setDimensionIfNotNull(
- DruidMetrics.TAGS,
- spec.getContextValue(DruidMetrics.TAGS)
- )
- .setMetric(
- AUTOSCALER_SCALING_TIME_METRIC,
- scaleActionStopwatch.millisElapsed()
- ));
- log.info("Changed taskCount to [%s] for supervisor[%s] for dataSource[%s].", desiredActiveTaskCount, supervisorId, dataSource);
- return true;
}
+ log.info(
+ "Starting scale action, current active task count is [%d] and desired task count is [%d] for supervisor[%s] for dataSource[%s].",
+ currentActiveTaskCount,
+ desiredActiveTaskCount,
+ supervisorId,
+ dataSource
+ );
+ final Stopwatch scaleActionStopwatch = Stopwatch.createStarted();
+ gracefulShutdownInternal();
+ changeTaskCountInAutoScalerConfig(desiredActiveTaskCount);
+ clearAllocationInfo();
+ emitter.emit(ServiceMetricEvent.builder()
+ .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
+ .setDimension(DruidMetrics.DATASOURCE, dataSource)
+ .setDimension(DruidMetrics.STREAM, getIoConfig().getStream())
+ .setDimensionIfNotNull(
+ DruidMetrics.TAGS,
+ spec.getContextValue(DruidMetrics.TAGS)
+ )
+ .setMetric(
+ AUTOSCALER_SCALING_TIME_METRIC,
+ scaleActionStopwatch.millisElapsed()
+ ));
+ log.info("Changed taskCount to [%s] for supervisor[%s] for dataSource[%s].", desiredActiveTaskCount, supervisorId, dataSource);
+ return true;
}
- private void changeTaskCountInIOConfig(int desiredActiveTaskCount)
+ private void changeTaskCountInAutoScalerConfig(int desiredActiveTaskCount)
{
+ // Sanity check.
+ if (autoScalerConfig == null) {
+ log.warn("autoScalerConfig is null but scale action is submitted, how can it be ?");
+ return;
+ }
Review Comment:
Please revert this change as well since the old method name is still relevant.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -255,6 +261,33 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
}
}
+ @Override
+ public void merge(@NotNull SupervisorSpec existingSpec)
+ {
+ AutoScalerConfig thisAutoScalerConfig = this.getIoConfig().getAutoScalerConfig();
+ // Either if autoscaler is absent or taskCountStart is specified - just return.
+ if (thisAutoScalerConfig == null || thisAutoScalerConfig.getTaskCountStart() != null) {
+ return;
+ }
Review Comment:
Should we also return early if `this.ioConfig.getTaskCount()` is specified?
##########
extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java:
##########
@@ -389,7 +389,7 @@ public SeekableStreamIndexTaskClient<KafkaTopicPartition, Long> build(
supervisor.start();
int taskCountBeforeScale = supervisor.getIoConfig().getTaskCount();
- Assert.assertEquals(1, taskCountBeforeScale);
+ Assert.assertNull(supervisor.getIoConfig().getAutoScalerConfig().getTaskCountStart());
Review Comment:
Please revert the changes to `KafkaSupervisorTest` and `KinesisSupervisorTest`.
I don't think they are needed anymore.
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SupervisorResourceConfigMergeTest.java:
##########
@@ -0,0 +1,533 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
+import org.apache.druid.audit.AuditManager;
+import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
+import org.apache.druid.indexing.overlord.TaskMaster;
+import org.apache.druid.indexing.overlord.TaskStorage;
+import org.apache.druid.indexing.overlord.supervisor.Supervisor;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorResource;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
+import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIOConfig;
+import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIngestionSpec;
+import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec;
+import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.AutoScalerConfig;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.metrics.DruidMonitorSchedulerConfig;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.server.security.Access;
+import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthConfig;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.Authorizer;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.Resource;
+import org.apache.druid.server.security.ResourceAction;
+import org.apache.druid.server.security.ResourceType;
+import org.easymock.Capture;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockRunner;
+import org.easymock.EasyMockSupport;
+import org.easymock.Mock;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.core.Response;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Set;
+
+@RunWith(EasyMockRunner.class)
+public class SupervisorResourceConfigMergeTest extends EasyMockSupport
+{
+ private static final ObjectMapper OBJECT_MAPPER = TestHelper.makeJsonMapper();
+ @Mock
+ private TaskMaster taskMaster;
+
+ @Mock
+ private SupervisorManager supervisorManager;
+
+ @Mock
+ private HttpServletRequest request;
+
+ @Mock
+ private AuthConfig authConfig;
+
+ @Mock
+ private AuditManager auditManager;
+
+ private SupervisorResource supervisorResource;
+
+ @Before
+ public void setUp()
+ {
+ supervisorResource = new SupervisorResource(
+ taskMaster,
+ new AuthorizerMapper(null)
+ {
+ @Override
+ public Authorizer getAuthorizer(String name)
+ {
+ return (authenticationResult, resource, action) -> {
+ if (authenticationResult.getIdentity().equals("druid")) {
+ return Access.OK;
+ } else {
+ if (resource.getType().equals(ResourceType.DATASOURCE)) {
+ if (resource.getName().equals("datasource2")) {
+ return Access.deny("not authorized.");
+ } else {
+ return Access.OK;
+ }
+ } else if (resource.getType().equals(ResourceType.EXTERNAL)) {
+ if (resource.getName().equals("test")) {
+ return Access.deny("not authorized.");
+ } else {
+ return Access.OK;
+ }
+ }
+ return Access.OK;
+ }
+ };
+ }
+ },
+ OBJECT_MAPPER,
+ authConfig,
+ auditManager
+ );
+ }
+
+ /**
+ * Tests that when a new spec is submitted without taskCountStart, the merge function
+ * follows the priority: provided taskCountStart > provided taskCount > existing taskCount > provided taskCountMin.
+ * In this test, existing spec has taskCount=5, new spec has no taskCount set,
+ * so merge should use existing taskCount (5).
+ */
Review Comment:
Can we remove the javadocs and update the method name to include the expectation/condition?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -183,15 +185,16 @@ public DruidMonitorSchedulerConfig getMonitorSchedulerConfig()
/**
* An autoScaler instance will be returned depending on the autoScalerConfig. In case autoScalerConfig is null or autoScaler is disabled then NoopTaskAutoScaler will be returned.
+ *
* @param supervisor
* @return autoScaler
*/
@Override
public SupervisorTaskAutoScaler createAutoscaler(Supervisor supervisor)
{
- AutoScalerConfig autoScalerConfig = ingestionSchema.getIOConfig().getAutoScalerConfig();
- if (autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler() && supervisor instanceof SeekableStreamSupervisor) {
- return autoScalerConfig.createAutoScaler(supervisor, this, emitter);
+ AutoScalerConfig autoScalerCfg = ingestionSchema.getIOConfig().getAutoScalerConfig();
Review Comment:
Please revert this rename as it does not seem required.
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SupervisorResourceConfigMergeTest.java:
##########
@@ -0,0 +1,533 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.seekablestream;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
+import org.apache.druid.audit.AuditManager;
+import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
+import org.apache.druid.indexing.overlord.TaskMaster;
+import org.apache.druid.indexing.overlord.TaskStorage;
+import org.apache.druid.indexing.overlord.supervisor.Supervisor;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorResource;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorSpec;
+import org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
+import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIOConfig;
+import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIngestionSpec;
+import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec;
+import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.AutoScalerConfig;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.java.util.metrics.DruidMonitorSchedulerConfig;
+import org.apache.druid.segment.TestHelper;
+import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.server.security.Access;
+import org.apache.druid.server.security.Action;
+import org.apache.druid.server.security.AuthConfig;
+import org.apache.druid.server.security.AuthenticationResult;
+import org.apache.druid.server.security.Authorizer;
+import org.apache.druid.server.security.AuthorizerMapper;
+import org.apache.druid.server.security.Resource;
+import org.apache.druid.server.security.ResourceAction;
+import org.apache.druid.server.security.ResourceType;
+import org.easymock.Capture;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockRunner;
+import org.easymock.EasyMockSupport;
+import org.easymock.Mock;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import javax.servlet.http.HttpServletRequest;
+import javax.ws.rs.core.Response;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Set;
+
+@RunWith(EasyMockRunner.class)
+public class SupervisorResourceConfigMergeTest extends EasyMockSupport
Review Comment:
Please move the test methods in this class to `SupervisorResourceTest` itself.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -556,44 +566,52 @@ public String getType()
private boolean changeTaskCount(int desiredActiveTaskCount)
throws InterruptedException, ExecutionException
{
+ if (autoScalerConfig == null) {
+ log.warn("autoScalerConfig is 'null' but dynamic allocation notice is submitted, how can it be ?");
+ return false;
+ }
int currentActiveTaskCount;
Collection<TaskGroup> activeTaskGroups = activelyReadingTaskGroups.values();
currentActiveTaskCount = activeTaskGroups.size();
if (desiredActiveTaskCount < 0 || desiredActiveTaskCount == currentActiveTaskCount) {
return false;
- } else {
- log.info(
- "Starting scale action, current active task count is [%d] and desired task count is [%d] for supervisor[%s] for dataSource[%s].",
- currentActiveTaskCount,
- desiredActiveTaskCount,
- supervisorId,
- dataSource
- );
- final Stopwatch scaleActionStopwatch = Stopwatch.createStarted();
- gracefulShutdownInternal();
- changeTaskCountInIOConfig(desiredActiveTaskCount);
- clearAllocationInfo();
- emitter.emit(ServiceMetricEvent.builder()
- .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
- .setDimension(DruidMetrics.DATASOURCE, dataSource)
- .setDimension(DruidMetrics.STREAM, getIoConfig().getStream())
- .setDimensionIfNotNull(
- DruidMetrics.TAGS,
- spec.getContextValue(DruidMetrics.TAGS)
- )
- .setMetric(
- AUTOSCALER_SCALING_TIME_METRIC,
- scaleActionStopwatch.millisElapsed()
- ));
- log.info("Changed taskCount to [%s] for supervisor[%s] for dataSource[%s].", desiredActiveTaskCount, supervisorId, dataSource);
- return true;
}
+ log.info(
+ "Starting scale action, current active task count is [%d] and desired task count is [%d] for supervisor[%s] for dataSource[%s].",
+ currentActiveTaskCount,
+ desiredActiveTaskCount,
+ supervisorId,
+ dataSource
+ );
+ final Stopwatch scaleActionStopwatch = Stopwatch.createStarted();
+ gracefulShutdownInternal();
+ changeTaskCountInAutoScalerConfig(desiredActiveTaskCount);
+ clearAllocationInfo();
+ emitter.emit(ServiceMetricEvent.builder()
+ .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
+ .setDimension(DruidMetrics.DATASOURCE, dataSource)
+ .setDimension(DruidMetrics.STREAM, getIoConfig().getStream())
+ .setDimensionIfNotNull(
+ DruidMetrics.TAGS,
+ spec.getContextValue(DruidMetrics.TAGS)
+ )
+ .setMetric(
+ AUTOSCALER_SCALING_TIME_METRIC,
+ scaleActionStopwatch.millisElapsed()
+ ));
+ log.info("Changed taskCount to [%s] for supervisor[%s] for dataSource[%s].", desiredActiveTaskCount, supervisorId, dataSource);
+ return true;
Review Comment:
Please revert this change to avoid bloating the diff.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3631029698
@Fly-Style , I think @jtuglu1 makes a fair point. I too would prefer that `taskCountStart` remain immutable and we retain the capability to reset the task count upon resubmission.
How about the following approach instead?
I have slightly different suggestion than the one captured in the PR description:
- [x] When submitting a supervisor with auto-scaler disabled, use the `taskCount` as is.
- [ ] When submitting a (new or existing) supervisor with auto-scaler enabled, set `taskCount = taskCountStart == null ? taskCountMin : taskCountStart`.
Persist the supervisor.
- [x] For any auto-scaling event, update the `taskCount` the same way that we do today.
- [ ] When the supervisor starts, it should start with the `taskCount` as the starting number of tasks rather than `taskCountStart`.
This would ensure:
- the issue with Overlord restarts is fixed
- `taskCountStart` remains an immutable config
- `taskCount` always reflects the current task count in the supervisor
What do you think?
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3533853597
Similarly, have you tested what happens when a supervisor is terminated (tombstoned) and then resubmitted without a taskCountStart? It'd be good to make sure that we don't end up merging an old supervisor (with same ID) data with potentially unrelated, new supervisor's data.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18745:
URL: https://github.com/apache/druid/pull/18745#discussion_r2605151958
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/AutoScalerConfig.java:
##########
@@ -40,6 +40,7 @@ public interface AutoScalerConfig
int getTaskCountMax();
int getTaskCountMin();
Integer getTaskCountStart();
+ void setTaskCountStart(int taskCountStart);
Review Comment:
This may not be needed anymore.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "Fly-Style (via GitHub)" <gi...@apache.org>.
Fly-Style commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3632368988
> let's try to add an embedded test for LagBasedAutoScaler similar to what you are doing in the other PR for cost-based auto-scaler.
Would love to, but in separate PR if you don't mind :)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3632293049
> Similarly, have you tested what happens when a supervisor is terminated (tombstoned) and then resubmitted without a taskCountStart? It'd be good to make sure that we don't end up merging an old supervisor (with same ID) data with potentially unrelated, new supervisor's data.
@jtuglu1 , this is already handled since the `SupervisorManager` contains only the non-tombstoned latest supervisor versions in memory. But we could add an embedded test to verify the same.
@Fly-Style , let's try to add an embedded test for `LagBasedAutoScaler` similar to what you are doing in the other PR for cost-based auto-scaler. The test could have a method which verifies that we do not pick up the `taskCountStart` from tombstoned versions, even if the supervisor id is the same.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3633901364
> I'm open to some way of offering both behaviors. I personally feel like sticking to the current task count on resubmitting is more intuitive as a default (since it seems odd that changing schema would lead to a reset of the task count). But I can live with the current behavior as default, as long as there's some way to get the stick-to-current behavior. Currently, there isn't.
A possibility:
- When a supervisor is started through `supervisor.start()`, initial task count is always taken from `taskCount`. This keeps task count consistent through Overlord restarts, etc.
- When a supervisor is posted through `SupervisorResource#specPost`, `taskCount` is set to:
1. `taskCountStart` if that is nonnull
2. else, the user-provided `taskCount` if that is nonnull
3. else, the `taskCount` from the previous supervisor spec (in the DB) if one exists
4. else, the user-provided `taskCountMin`
This I think allows @jtuglu1 and me to both have the behavior we want: @jtuglu1 would set `taskCountStart` and the task count would always reset to that. I would set neither `taskCount` nor `taskCountStart`, and the task count would start out at `taskCountMin` for a fresh supervisor, then retain the current count from then on.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #18745:
URL: https://github.com/apache/druid/pull/18745#discussion_r2526861904
##########
indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamSupervisorSpecTest.java:
##########
@@ -566,14 +566,16 @@
Exception e = null;
try {
- AutoScalerConfig autoScalerError = mapper.convertValue(ImmutableMap.of(
- "enableTaskAutoScaler",
- "true",
- "taskCountMax",
- "1",
- "taskCountMin",
- "4"
- ), AutoScalerConfig.class);
+ AutoScalerConfig autoScalerError = mapper.convertValue(
+ ImmutableMap.of(
+ "enableTaskAutoScaler",
+ "true",
+ "taskCountMax",
+ "1",
+ "taskCountMin",
+ "4"
+ ), AutoScalerConfig.class
+ );
Review Comment:
## Unread local variable
Variable 'AutoScalerConfig autoScalerError' is never read.
[Show more details](https://github.com/apache/druid/security/code-scanning/10554)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "Fly-Style (via GitHub)" <gi...@apache.org>.
Fly-Style commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3635886205
@gianm, @kfaraz, @jtuglu1 thanks a lot for a productive discussion! Happy to implement this according to all requests!
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3635335511
SGTM
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "Fly-Style (via GitHub)" <gi...@apache.org>.
Fly-Style commented on code in PR #18745:
URL: https://github.com/apache/druid/pull/18745#discussion_r2611305867
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -255,6 +261,33 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
}
}
+ @Override
+ public void merge(@NotNull SupervisorSpec existingSpec)
+ {
+ AutoScalerConfig thisAutoScalerConfig = this.getIoConfig().getAutoScalerConfig();
+ // Either if autoscaler is absent or taskCountStart is specified - just return.
+ if (thisAutoScalerConfig == null || thisAutoScalerConfig.getTaskCountStart() != null) {
+ return;
+ }
Review Comment:
Not really, taskCountStart has bigger priority :)
> the priority: provided `taskCountStart` > provided `taskCount` > existing `taskCount` > provided `taskCountMin`.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz merged PR #18745:
URL: https://github.com/apache/druid/pull/18745
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "Fly-Style (via GitHub)" <gi...@apache.org>.
Fly-Style commented on code in PR #18745:
URL: https://github.com/apache/druid/pull/18745#discussion_r2602756224
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -556,44 +566,52 @@ public String getType()
private boolean changeTaskCount(int desiredActiveTaskCount)
throws InterruptedException, ExecutionException
{
+ if (autoScalerConfig == null) {
+ log.warn("autoScalerConfig is 'null' but dynamic allocation notice is submitted, how can it be ?");
+ return false;
+ }
int currentActiveTaskCount;
Collection<TaskGroup> activeTaskGroups = activelyReadingTaskGroups.values();
currentActiveTaskCount = activeTaskGroups.size();
if (desiredActiveTaskCount < 0 || desiredActiveTaskCount == currentActiveTaskCount) {
return false;
- } else {
- log.info(
- "Starting scale action, current active task count is [%d] and desired task count is [%d] for supervisor[%s] for dataSource[%s].",
- currentActiveTaskCount,
- desiredActiveTaskCount,
- supervisorId,
- dataSource
- );
- final Stopwatch scaleActionStopwatch = Stopwatch.createStarted();
- gracefulShutdownInternal();
- changeTaskCountInIOConfig(desiredActiveTaskCount);
- clearAllocationInfo();
- emitter.emit(ServiceMetricEvent.builder()
- .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
- .setDimension(DruidMetrics.DATASOURCE, dataSource)
- .setDimension(DruidMetrics.STREAM, getIoConfig().getStream())
- .setDimensionIfNotNull(
- DruidMetrics.TAGS,
- spec.getContextValue(DruidMetrics.TAGS)
- )
- .setMetric(
- AUTOSCALER_SCALING_TIME_METRIC,
- scaleActionStopwatch.millisElapsed()
- ));
- log.info("Changed taskCount to [%s] for supervisor[%s] for dataSource[%s].", desiredActiveTaskCount, supervisorId, dataSource);
- return true;
}
+ log.info(
+ "Starting scale action, current active task count is [%d] and desired task count is [%d] for supervisor[%s] for dataSource[%s].",
+ currentActiveTaskCount,
+ desiredActiveTaskCount,
+ supervisorId,
+ dataSource
+ );
+ final Stopwatch scaleActionStopwatch = Stopwatch.createStarted();
+ gracefulShutdownInternal();
+ changeTaskCountInAutoScalerConfig(desiredActiveTaskCount);
+ clearAllocationInfo();
+ emitter.emit(ServiceMetricEvent.builder()
+ .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
+ .setDimension(DruidMetrics.DATASOURCE, dataSource)
+ .setDimension(DruidMetrics.STREAM, getIoConfig().getStream())
+ .setDimensionIfNotNull(
+ DruidMetrics.TAGS,
+ spec.getContextValue(DruidMetrics.TAGS)
+ )
+ .setMetric(
+ AUTOSCALER_SCALING_TIME_METRIC,
+ scaleActionStopwatch.millisElapsed()
+ ));
+ log.info("Changed taskCount to [%s] for supervisor[%s] for dataSource[%s].", desiredActiveTaskCount, supervisorId, dataSource);
+ return true;
Review Comment:
Yeap, it is obsolete in my taste.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "gianm (via GitHub)" <gi...@apache.org>.
gianm commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3631424435
At our deployment, we do want supervisors to stick to the current task count when resubmitted. The reasons are:
- Supervisors need to be updated periodically to do stuff like change tuning configs, change schemas, etc. When we do a change like this, we don't want the task count to change from whatever is currently running.
- With the way we use autoscaling, the `taskCountStart` is just some fixed starter value that is the same for all tables, not anything actually related to that particular table. We rely on the autoscaler to find the ideal value. So, reverting back to the `taskCountStart` is not desirable.
I'm open to some way of offering both behaviors. I personally feel like sticking to the current task count on resubmitting is more intuitive as a default (since it seems odd that changing schema would lead to a reset of the task count). But I can live with the current behavior as default, as long as there's some way to get the stick-to-current behavior. Currently, there isn't.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "jtuglu1 (via GitHub)" <gi...@apache.org>.
jtuglu1 commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3533773873
> When the autoscaler does a scaling action, it updates ioConfig.autoScalerConfig.taskCountStart, not ioConfig.taskCount
Hi, @Fly-Style thanks for the change. Not sure the above makes sense to me. `autoScalerConfig.taskCountStart` is supposed to be an immutable value used on supervisor submit to allow the scaler to not have to start at minimum task count (this makes having a smaller minimum easier). Scaling actions should not touch this value. From the docs:
> Optional config to specify the number of ingestion tasks to start with. When you enable the autoscaler, Druid ignores the value of taskCount in ioConfig and, if specified, starts with the taskCountStart number of tasks. Otherwise, defaults to taskCountMin.
`ioConfig.taskCount` is the currently running taskCount and should be what is updated.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on code in PR #18745:
URL: https://github.com/apache/druid/pull/18745#discussion_r2602627218
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1339,7 +1357,7 @@ public void tryInit()
}
initRetryCounter++;
log.makeAlert(e, "Exception starting SeekableStreamSupervisor[%s]", supervisorId)
- .emit();
+ .emit();
Review Comment:
not needed?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/LagBasedAutoScaler.java:
##########
@@ -190,16 +190,22 @@ private Runnable computeAndCollectLag()
};
}
- /**
- * This method determines whether to do scale actions based on collected lag points.
- * Current algorithm of scale is simple:
- * First of all, compute the proportion of lag points higher/lower than scaleOutThreshold/scaleInThreshold, getting scaleOutThreshold/scaleInThreshold.
- * Secondly, compare scaleOutThreshold/scaleInThreshold with triggerScaleOutFractionThreshold/triggerScaleInFractionThreshold. P.S. Scale out action has higher priority than scale in action.
- * Finaly, if scaleOutThreshold/scaleInThreshold is higher than triggerScaleOutFractionThreshold/triggerScaleInFractionThreshold, scale out/in action would be triggered.
- *
- * @param lags the lag metrics of Stream(Kafka/Kinesis)
- * @return Integer. target number of tasksCount, -1 means skip scale action.
- */
+ /**
+ * This method determines whether to do scale actions based on collected lag points.
+ * The current algorithm of scale is straightforward:
+ * <ul>
+ * <li>First, compute the proportion of lag points higher/lower than <code>scaleOutThreshold</code>/<code>scaleInThreshold</code>,
Review Comment:
You can also use the `{@code}` javadoc tag as it is less verbose.
```suggestion
* <li>First, compute the proportion of lag points higher/lower than {@code scaleOutThreshold/scaleInThreshold},
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1161,7 +1179,7 @@ public void stop(boolean stopGracefully)
catch (Exception e) {
stateManager.recordThrowableEvent(e);
log.makeAlert(e, "Exception stopping [%s]", supervisorId)
- .emit();
+ .emit();
Review Comment:
not needed?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -255,6 +263,32 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
}
}
+ /**
+ * Writes the <code>taskCountStart</code> value from old config, if not specificed in new config.
+ *
+ * @param existingSpec the existing supervisor specification to merge configuration values from
+ */
+ @Override
+ public void mergeSpecConfigs(@NotNull SupervisorSpec existingSpec)
+ {
+ AutoScalerConfig thisAutoScalerConfig = this.getIoConfig().getAutoScalerConfig();
+ // Either if autoscaler is absent or taskCountStart is specified - just return.
+ if (thisAutoScalerConfig == null || thisAutoScalerConfig.getTaskCountStart() != null) {
+ return;
+ }
+
+ // TODO[sasha]: use switch expression with pattern matching when we move to Java 21 as minimum requirement.
+ if (existingSpec instanceof SeekableStreamSupervisorSpec) {
+ // Note: for some reason, sources are available only for bytecode version 11.
Review Comment:
Please remove this.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -255,6 +263,32 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
}
}
+ /**
+ * Writes the <code>taskCountStart</code> value from old config, if not specificed in new config.
+ *
+ * @param existingSpec the existing supervisor specification to merge configuration values from
+ */
+ @Override
+ public void mergeSpecConfigs(@NotNull SupervisorSpec existingSpec)
+ {
+ AutoScalerConfig thisAutoScalerConfig = this.getIoConfig().getAutoScalerConfig();
+ // Either if autoscaler is absent or taskCountStart is specified - just return.
+ if (thisAutoScalerConfig == null || thisAutoScalerConfig.getTaskCountStart() != null) {
+ return;
+ }
+
+ // TODO[sasha]: use switch expression with pattern matching when we move to Java 21 as minimum requirement.
Review Comment:
```suggestion
// Use switch expression with pattern matching when we move to Java 21 as minimum requirement.
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -255,6 +263,32 @@ public void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcept
}
}
+ /**
+ * Writes the <code>taskCountStart</code> value from old config, if not specificed in new config.
+ *
+ * @param existingSpec the existing supervisor specification to merge configuration values from
+ */
+ @Override
+ public void mergeSpecConfigs(@NotNull SupervisorSpec existingSpec)
+ {
+ AutoScalerConfig thisAutoScalerConfig = this.getIoConfig().getAutoScalerConfig();
+ // Either if autoscaler is absent or taskCountStart is specified - just return.
+ if (thisAutoScalerConfig == null || thisAutoScalerConfig.getTaskCountStart() != null) {
+ return;
+ }
+
+ // TODO[sasha]: use switch expression with pattern matching when we move to Java 21 as minimum requirement.
+ if (existingSpec instanceof SeekableStreamSupervisorSpec) {
+ // Note: for some reason, sources are available only for bytecode version 11.
+ //noinspection PatternVariableCanBeUsed
+ var spec = (SeekableStreamSupervisorSpec) existingSpec;
Review Comment:
```suggestion
SeekableStreamSupervisorSpec spec = (SeekableStreamSupervisorSpec) existingSpec;
```
##########
server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpec.java:
##########
@@ -113,4 +114,15 @@ default void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcep
{
// The default implementation does not do any validation checks.
}
+
+ /**
+ * Perform any merging of spec configurations needed after deserialization.
Review Comment:
```suggestion
* Updates this supervisor spec by merging values from the given {@code existingSpec}.
* This method may be used to carry forward existing spec values when a supervisor is being resubmitted.
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -556,44 +566,52 @@ public String getType()
private boolean changeTaskCount(int desiredActiveTaskCount)
throws InterruptedException, ExecutionException
{
+ if (autoScalerConfig == null) {
+ log.warn("autoScalerConfig is 'null' but dynamic allocation notice is submitted, how can it be ?");
+ return false;
+ }
int currentActiveTaskCount;
Collection<TaskGroup> activeTaskGroups = activelyReadingTaskGroups.values();
currentActiveTaskCount = activeTaskGroups.size();
if (desiredActiveTaskCount < 0 || desiredActiveTaskCount == currentActiveTaskCount) {
return false;
- } else {
- log.info(
- "Starting scale action, current active task count is [%d] and desired task count is [%d] for supervisor[%s] for dataSource[%s].",
- currentActiveTaskCount,
- desiredActiveTaskCount,
- supervisorId,
- dataSource
- );
- final Stopwatch scaleActionStopwatch = Stopwatch.createStarted();
- gracefulShutdownInternal();
- changeTaskCountInIOConfig(desiredActiveTaskCount);
- clearAllocationInfo();
- emitter.emit(ServiceMetricEvent.builder()
- .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
- .setDimension(DruidMetrics.DATASOURCE, dataSource)
- .setDimension(DruidMetrics.STREAM, getIoConfig().getStream())
- .setDimensionIfNotNull(
- DruidMetrics.TAGS,
- spec.getContextValue(DruidMetrics.TAGS)
- )
- .setMetric(
- AUTOSCALER_SCALING_TIME_METRIC,
- scaleActionStopwatch.millisElapsed()
- ));
- log.info("Changed taskCount to [%s] for supervisor[%s] for dataSource[%s].", desiredActiveTaskCount, supervisorId, dataSource);
- return true;
}
+ log.info(
+ "Starting scale action, current active task count is [%d] and desired task count is [%d] for supervisor[%s] for dataSource[%s].",
+ currentActiveTaskCount,
+ desiredActiveTaskCount,
+ supervisorId,
+ dataSource
+ );
+ final Stopwatch scaleActionStopwatch = Stopwatch.createStarted();
+ gracefulShutdownInternal();
+ changeTaskCountInAutoScalerConfig(desiredActiveTaskCount);
+ clearAllocationInfo();
+ emitter.emit(ServiceMetricEvent.builder()
+ .setDimension(DruidMetrics.SUPERVISOR_ID, supervisorId)
+ .setDimension(DruidMetrics.DATASOURCE, dataSource)
+ .setDimension(DruidMetrics.STREAM, getIoConfig().getStream())
+ .setDimensionIfNotNull(
+ DruidMetrics.TAGS,
+ spec.getContextValue(DruidMetrics.TAGS)
+ )
+ .setMetric(
+ AUTOSCALER_SCALING_TIME_METRIC,
+ scaleActionStopwatch.millisElapsed()
+ ));
+ log.info("Changed taskCount to [%s] for supervisor[%s] for dataSource[%s].", desiredActiveTaskCount, supervisorId, dataSource);
+ return true;
Review Comment:
I guess this is just moving the code out of the `else` block. Can we revert this change for the time being?
##########
server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpec.java:
##########
@@ -113,4 +114,15 @@ default void validateSpecUpdateTo(SupervisorSpec proposedSpec) throws DruidExcep
{
// The default implementation does not do any validation checks.
}
+
+ /**
+ * Perform any merging of spec configurations needed after deserialization.
+ *
+ * @param existingSpec used spec to merge values from
+ * @throws DruidException if the spec update is not allowed
+ */
+ default void mergeSpecConfigs(@NotNull SupervisorSpec existingSpec)
Review Comment:
```suggestion
default void merge(@NotNull SupervisorSpec existingSpec)
```
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Improve lag-based autoscaler config persistence (druid)
Posted by "kfaraz (via GitHub)" <gi...@apache.org>.
kfaraz commented on PR #18745:
URL: https://github.com/apache/druid/pull/18745#issuecomment-3635332606
Thanks for the suggestion, @gianm ! I think that would work nicely and meet all our needs.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org