This is an automated email from the ASF dual-hosted git repository.
Fly-Style pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new f0d8c616cc5 feat: rework `requireRestart` to `getActionOnUpdateTo` to
control restart/task-disruption behavior (#19700)
f0d8c616cc5 is described below
commit f0d8c616cc577f39da8db98415186264b868767c
Author: Sasha Syrotenko <[email protected]>
AuthorDate: Mon Jul 20 16:13:04 2026 +0300
feat: rework `requireRestart` to `getActionOnUpdateTo` to control
restart/task-disruption behavior (#19700)
Today, updating a supervisor spec always restarts the supervisor and
terminates its running tasks, even when the change is cosmetic. This makes
routine spec updates unnecessarily disruptive to in-flight ingestion.
This PR replaces the boolean `SupervisorSpec#requireRestart` with a
tri-state `SupervisorSpec#getActionOnUpdateTo(SupervisorSpec proposedSpec`),
returning a `SupervisorSpecUpdateAction`:
- `NONE` - persist the new spec, no restart
- `RESTART_SUPERVISOR` - restart the supervisor, but leave running tasks
alone
- `RESTART_SUPERVISOR_AND_TASKS` - restart the supervisor and terminate
its tasks (previous behavior; still the default)
---
.../RabbitStreamSupervisorTuningConfigTest.java | 18 +++++++---
.../overlord/supervisor/SupervisorManager.java | 39 +++++++++++++---------
.../supervisor/SupervisorSpecUpdateResult.java | 13 ++++----
.../supervisor/SeekableStreamSupervisorSpec.java | 11 +++---
.../http/OverlordCompactionResourceTest.java | 3 +-
.../overlord/supervisor/SupervisorManagerTest.java | 11 +++---
.../supervisor/SupervisorResourceTest.java | 12 +++----
.../SeekableStreamSupervisorSpecTest.java | 22 +++++++-----
.../overlord/supervisor/SupervisorSpec.java | 10 +++---
.../supervisor/SupervisorSpecUpdateAction.java | 31 ++++-------------
10 files changed, 90 insertions(+), 80 deletions(-)
diff --git
a/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTuningConfigTest.java
b/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTuningConfigTest.java
index f4846573d3e..2fa6fc4ee7e 100644
---
a/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTuningConfigTest.java
+++
b/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTuningConfigTest.java
@@ -24,6 +24,7 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.druid.data.input.impl.DimensionsSpec;
import org.apache.druid.data.input.impl.TimestampSpec;
import org.apache.druid.indexer.granularity.UniformGranularitySpec;
+import
org.apache.druid.indexing.overlord.supervisor.SupervisorSpecUpdateAction;
import org.apache.druid.indexing.rabbitstream.RabbitStreamIndexTaskModule;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.granularity.Granularities;
@@ -59,10 +60,19 @@ public class RabbitStreamSupervisorTuningConfigTest
{
final RabbitStreamSupervisorSpec oldSpec = supervisorSpec(tuningConfig(15,
16, 17));
- // requireRestart is invoked on the running (old) spec with the proposed
spec as argument.
- Assert.assertTrue(oldSpec.requireRestart(supervisorSpec(tuningConfig(20,
16, 17))));
- Assert.assertTrue(oldSpec.requireRestart(supervisorSpec(tuningConfig(15,
20, 17))));
- Assert.assertTrue(oldSpec.requireRestart(supervisorSpec(tuningConfig(15,
16, 20))));
+ // getActionOnUpdateTo is invoked on the running (old) spec with the
proposed spec as argument.
+ Assert.assertEquals(
+ SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
+ oldSpec.getActionOnUpdateTo(supervisorSpec(tuningConfig(20, 16, 17)))
+ );
+ Assert.assertEquals(
+ SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
+ oldSpec.getActionOnUpdateTo(supervisorSpec(tuningConfig(15, 20, 17)))
+ );
+ Assert.assertEquals(
+ SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
+ oldSpec.getActionOnUpdateTo(supervisorSpec(tuningConfig(15, 16, 20)))
+ );
}
@Test
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
index 9f599f3aee8..7c75e5a9642 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java
@@ -178,8 +178,9 @@ public class SupervisorManager implements
SupervisorStatsProvider
/**
* Applies {@code spec} under a single lock. With {@code
skipRestartIfUnmodified=true}, an unchanged spec
- * is a no-op; a changed spec is persisted without restart when {@link
SupervisorSpec#requireRestart} is
- * false. With {@code skipRestartIfUnmodified=false}, the supervisor is
always recreated (legacy behavior).
+ * is a no-op; a changed spec is persisted without restart when {@link
SupervisorSpec#getActionOnUpdateTo}
+ * returns {@link SupervisorSpecUpdateAction#NONE}. With {@code
skipRestartIfUnmodified=false}, the supervisor
+ * is always recreated and its tasks always terminated.
*/
public SupervisorSpecUpdateResult createOrUpdateAndStartSupervisor(
final SupervisorSpec spec,
@@ -195,12 +196,12 @@ public class SupervisorManager implements
SupervisorStatsProvider
Preconditions.checkState(started, "SupervisorManager not started");
if (!skipRestartIfUnmodified) {
- // Always stop/recreate, persisting whenever the spec actually changed
(or is new).
+ // Always stop/recreate and terminate tasks, persisting whenever the
spec actually changed
final boolean specChanged = isSpecChangedAndValidated(spec);
- final SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ final SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false, true);
spec.merge(existingSpec);
createAndStartSupervisorInternal(spec, specChanged);
- return SupervisorSpecUpdateResult.of(specChanged, true);
+ return SupervisorSpecUpdateResult.of(specChanged,
SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS);
}
final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
@@ -210,33 +211,35 @@ public class SupervisorManager implements
SupervisorStatsProvider
// merge() self-initializes taskCount from taskCountStart for a new
autoscaling supervisor (no existing spec).
spec.merge(null);
createAndStartSupervisorInternal(spec, true);
- return SupervisorSpecUpdateResult.of(true, true);
+ return SupervisorSpecUpdateResult.of(true,
SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS);
}
final SupervisorSpec currentSpec = current.rhs;
if (!areSpecBytesChanged(spec, currentSpec)) {
- return SupervisorSpecUpdateResult.of(false, false);
+ return SupervisorSpecUpdateResult.of(false,
SupervisorSpecUpdateAction.NONE);
}
// merge() may carry forward omitted fields (e.g. taskCount); compare
the effective spec.
spec.merge(currentSpec);
if (!areSpecBytesChanged(spec, currentSpec)) {
- return SupervisorSpecUpdateResult.of(false, false);
+ return SupervisorSpecUpdateResult.of(false,
SupervisorSpecUpdateAction.NONE);
}
// The effective (merged) spec is what will be persisted, so validate
that transition exactly once.
currentSpec.validateSpecUpdateTo(spec);
- if (!currentSpec.requireRestart(spec)) {
+ SupervisorSpecUpdateAction action =
currentSpec.getActionOnUpdateTo(spec);
+ if (action == SupervisorSpecUpdateAction.NONE) {
metadataSupervisorManager.insert(spec.getId(), spec);
supervisors.put(spec.getId(), Pair.of(current.lhs, spec));
- return SupervisorSpecUpdateResult.of(true, false);
+ return SupervisorSpecUpdateResult.of(true,
SupervisorSpecUpdateAction.NONE);
}
// Restart path: stop+recreate, persisting the changed spec.
- possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ boolean shouldTaskBeTerminated = action ==
SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS;
+ possiblyStopAndRemoveSupervisorInternal(spec.getId(), false,
shouldTaskBeTerminated);
createAndStartSupervisorInternal(spec, true);
- return SupervisorSpecUpdateResult.of(true, true);
+ return SupervisorSpecUpdateResult.of(true, action);
}
}
@@ -277,7 +280,7 @@ public class SupervisorManager implements
SupervisorStatsProvider
synchronized (lock) {
Preconditions.checkState(started, "SupervisorManager not started");
- return possiblyStopAndRemoveSupervisorInternal(id, true) != null;
+ return possiblyStopAndRemoveSupervisorInternal(id, true, true) != null;
}
}
@@ -650,7 +653,11 @@ public class SupervisorManager implements
SupervisorStatsProvider
* @return reference to existing supervisor, if exists and was stopped, null
if there was no supervisor with this id
*/
@Nullable
- private SupervisorSpec possiblyStopAndRemoveSupervisorInternal(String id,
boolean writeTombstone)
+ private SupervisorSpec possiblyStopAndRemoveSupervisorInternal(
+ String id,
+ boolean writeTombstone,
+ boolean terminateTasks
+ )
{
Pair<Supervisor, SupervisorSpec> pair = supervisors.get(id);
if (pair == null || pair.rhs == null || pair.lhs == null) {
@@ -663,7 +670,7 @@ public class SupervisorManager implements
SupervisorStatsProvider
new NoopSupervisorSpec(null, pair.rhs.getDataSources())
); // where NoopSupervisorSpec is a tombstone
}
- pair.lhs.stop(true);
+ pair.lhs.stop(terminateTasks);
supervisors.remove(id);
SupervisorTaskAutoScaler autoscaler = autoscalers.get(id);
@@ -692,7 +699,7 @@ public class SupervisorManager implements
SupervisorStatsProvider
}
SupervisorSpec nextState = suspend ? pair.rhs.createSuspendedSpec() :
pair.rhs.createRunningSpec();
- possiblyStopAndRemoveSupervisorInternal(nextState.getId(), false);
+ possiblyStopAndRemoveSupervisorInternal(nextState.getId(), false, true);
return createAndStartSupervisorInternal(nextState, true);
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateResult.java
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateResult.java
index bf14917672c..39266a3b04e 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateResult.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateResult.java
@@ -25,17 +25,17 @@ package org.apache.druid.indexing.overlord.supervisor;
public final class SupervisorSpecUpdateResult
{
private final boolean modified;
- private final boolean restarted;
+ private final SupervisorSpecUpdateAction action;
- private SupervisorSpecUpdateResult(final boolean modified, final boolean
restarted)
+ private SupervisorSpecUpdateResult(final boolean modified, final
SupervisorSpecUpdateAction action)
{
this.modified = modified;
- this.restarted = restarted;
+ this.action = action;
}
- public static SupervisorSpecUpdateResult of(final boolean modified, final
boolean restarted)
+ public static SupervisorSpecUpdateResult of(final boolean modified, final
SupervisorSpecUpdateAction action)
{
- return new SupervisorSpecUpdateResult(modified, restarted);
+ return new SupervisorSpecUpdateResult(modified, action);
}
public boolean isModified()
@@ -45,6 +45,7 @@ public final class SupervisorSpecUpdateResult
public boolean isRestarted()
{
- return restarted;
+ return action != SupervisorSpecUpdateAction.NONE;
}
+
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java
index 6c7da51facb..5a94168102c 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java
+++
b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java
@@ -33,6 +33,7 @@ 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.SupervisorSpec;
+import
org.apache.druid.indexing.overlord.supervisor.SupervisorSpecUpdateAction;
import
org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
import
org.apache.druid.indexing.overlord.supervisor.autoscaler.SupervisorTaskAutoScaler;
import
org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskClientFactory;
@@ -298,10 +299,10 @@ public abstract class SeekableStreamSupervisorSpec
implements SupervisorSpec
}
@Override
- public boolean requireRestart(SupervisorSpec proposedSpec)
+ public SupervisorSpecUpdateAction getActionOnUpdateTo(SupervisorSpec
proposedSpec)
{
if (!(proposedSpec instanceof SeekableStreamSupervisorSpec proposed)) {
- return true;
+ return SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS;
}
final Builder<?> proposedCopy = proposed.toBuilder();
@@ -313,7 +314,9 @@ public abstract class SeekableStreamSupervisorSpec
implements SupervisorSpec
proposedCopy.taskCount(getIoConfig().getTaskCount());
}
- return !proposedCopy.build().equals(this);
+ return !proposedCopy.build().equals(this)
+ ? SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS
+ : SupervisorSpecUpdateAction.NONE;
}
private boolean isAutoScalerEnabled()
@@ -353,7 +356,7 @@ public abstract class SeekableStreamSupervisorSpec
implements SupervisorSpec
);
/**
- * Returns a copy builder seeded from this spec, used by {@link
#requireRestart} for structured comparison.
+ * Returns a copy builder seeded from this spec, used by {@link
#getActionOnUpdateTo} for structured comparison.
* Abstract so that every stream supervisor spec participates in the restart
decision.
*/
public abstract Builder<?> toBuilder();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java
index 940dd211542..53a5d05f3f1 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/http/OverlordCompactionResourceTest.java
@@ -31,6 +31,7 @@ import
org.apache.druid.indexing.compact.CompactionSupervisorSpec;
import org.apache.druid.indexing.overlord.TaskMaster;
import
org.apache.druid.indexing.overlord.supervisor.CompactionSupervisorManager;
import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
+import
org.apache.druid.indexing.overlord.supervisor.SupervisorSpecUpdateAction;
import
org.apache.druid.indexing.overlord.supervisor.SupervisorSpecUpdateResult;
import org.apache.druid.indexing.overlord.supervisor.VersionedSupervisorSpec;
import org.apache.druid.java.util.common.DateTimes;
@@ -317,7 +318,7 @@ public class OverlordCompactionResourceTest
new CompactionSupervisorSpec(wikiConfig, false, validator);
EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(supervisorSpec,
true))
- .andReturn(SupervisorSpecUpdateResult.of(true, true)).once();
+ .andReturn(SupervisorSpecUpdateResult.of(true,
SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS)).once();
EasyMock.expect(scheduler.validateCompactionConfig(wikiConfig))
.andReturn(CompactionConfigValidationResult.success()).once();
replayAll();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
index ec104eb2215..7fbb130d2b0 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java
@@ -1610,14 +1610,15 @@ public class SupervisorManagerTest extends
EasyMockSupport
}
/**
- * Lets a test model either a no-restart change (default) or a
restart-requiring change. Invoked on the
- * running spec, so this models whether the running supervisor requires a
restart. (The default
- * {@link SupervisorSpec#requireRestart} is conservative and returns true.)
+ * Lets a test model either a no-restart change (default) or a
restart-requiring change.
+ * Invoked on therunning spec, so this models whether the running
supervisor requires a restart.
+ * The default {@link SupervisorSpec#getActionOnUpdateTo} is conservative
and
+ * returns {@link SupervisorSpecUpdateAction#RESTART_SUPERVISOR_AND_TASKS}.
*/
@Override
- public boolean requireRestart(SupervisorSpec proposedSpec)
+ public SupervisorSpecUpdateAction getActionOnUpdateTo(SupervisorSpec
proposedSpec)
{
- return requireRestart;
+ return requireRestart ?
SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS :
SupervisorSpecUpdateAction.NONE;
}
}
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java
index 1ed1214368b..ff0f50d0b50 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java
@@ -165,7 +165,7 @@ public class SupervisorResourceTest extends EasyMockSupport
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec,
false))
- .andReturn(SupervisorSpecUpdateResult.of(true, true));
+ .andReturn(SupervisorSpecUpdateResult.of(true,
SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS));
setupMockRequest();
setupMockRequestForAudit();
@@ -185,7 +185,7 @@ public class SupervisorResourceTest extends EasyMockSupport
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec,
false))
- .andReturn(SupervisorSpecUpdateResult.of(false, true));
+ .andReturn(SupervisorSpecUpdateResult.of(false,
SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS));
setupMockRequest();
setupMockRequestForAudit();
@@ -261,7 +261,7 @@ public class SupervisorResourceTest extends EasyMockSupport
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
// Changed but no restart needed: persisted without restarting — and the
persist must be audited.
EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec,
true))
- .andReturn(SupervisorSpecUpdateResult.of(true, false));
+ .andReturn(SupervisorSpecUpdateResult.of(true,
SupervisorSpecUpdateAction.NONE));
setupMockRequest();
setupMockRequestForAudit();
@@ -281,7 +281,7 @@ public class SupervisorResourceTest extends EasyMockSupport
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec,
true))
- .andReturn(SupervisorSpecUpdateResult.of(false, false));
+ .andReturn(SupervisorSpecUpdateResult.of(false,
SupervisorSpecUpdateAction.NONE));
setupMockRequest();
@@ -299,7 +299,7 @@ public class SupervisorResourceTest extends EasyMockSupport
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec,
true))
- .andReturn(SupervisorSpecUpdateResult.of(true, true));
+ .andReturn(SupervisorSpecUpdateResult.of(true,
SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS));
setupMockRequest();
setupMockRequestForAudit();
@@ -332,7 +332,7 @@ public class SupervisorResourceTest extends EasyMockSupport
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec,
false))
- .andReturn(SupervisorSpecUpdateResult.of(true, true));
+ .andReturn(SupervisorSpecUpdateResult.of(true,
SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS));
setupMockRequest();
setupMockRequestForAudit();
diff --git
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
index 2b54f720e52..f5e203b0a00 100644
---
a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
+++
b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpecTest.java
@@ -29,6 +29,7 @@ import org.apache.druid.error.DruidExceptionMatcher;
import org.apache.druid.indexing.overlord.supervisor.NoopSupervisorSpec;
import org.apache.druid.indexing.overlord.supervisor.Supervisor;
import org.apache.druid.indexing.overlord.supervisor.SupervisorManager;
+import
org.apache.druid.indexing.overlord.supervisor.SupervisorSpecUpdateAction;
import org.apache.druid.indexing.overlord.supervisor.SupervisorStateManager;
import
org.apache.druid.indexing.overlord.supervisor.SupervisorStateManagerConfig;
import
org.apache.druid.indexing.overlord.supervisor.autoscaler.SupervisorTaskAutoScaler;
@@ -1430,7 +1431,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
buildSpecWithIoConfig("id", createIOConfig(2,
lagBasedAutoScalerConfig(1, 8, null)));
final SeekableStreamSupervisorSpec oldSpec = seed.toBuilder().build();
final SeekableStreamSupervisorSpec newSpec = seed.toBuilder().build();
- Assert.assertFalse(oldSpec.requireRestart(newSpec));
+ Assert.assertEquals(SupervisorSpecUpdateAction.NONE,
oldSpec.getActionOnUpdateTo(newSpec));
}
@Test
@@ -1440,7 +1441,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
buildSpecWithIoConfig("id", createIOConfig(2,
lagBasedAutoScalerConfig(1, 8, null)));
final SeekableStreamSupervisorSpec oldSpec =
seed.toBuilder().taskCount(2).build();
final SeekableStreamSupervisorSpec newSpec =
seed.toBuilder().taskCount(5).build();
- Assert.assertFalse(oldSpec.requireRestart(newSpec));
+ Assert.assertEquals(SupervisorSpecUpdateAction.NONE,
oldSpec.getActionOnUpdateTo(newSpec));
}
@Test
@@ -1449,7 +1450,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
final TestSeekableStreamSupervisorSpec seed = buildSpecWithIoConfig("id",
createIOConfig(2, null));
final SeekableStreamSupervisorSpec oldSpec =
seed.toBuilder().taskCount(2).build();
final SeekableStreamSupervisorSpec newSpec =
seed.toBuilder().taskCount(5).build();
- Assert.assertTrue(oldSpec.requireRestart(newSpec));
+
Assert.assertEquals(SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
oldSpec.getActionOnUpdateTo(newSpec));
}
@Test
@@ -1459,7 +1460,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
buildSpecWithIoConfig("id", createIOConfig(2,
lagBasedAutoScalerConfig(1, 8, null)));
final SeekableStreamSupervisorSpec oldSpec = seed.toBuilder().build();
final SeekableStreamSupervisorSpec newSpec =
seed.toBuilder().context(Map.of("k", "v")).build();
- Assert.assertTrue(oldSpec.requireRestart(newSpec));
+
Assert.assertEquals(SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
oldSpec.getActionOnUpdateTo(newSpec));
}
@Test
@@ -1469,7 +1470,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
buildSpecWithIoConfig("id", createIOConfig(2,
lagBasedAutoScalerConfig(1, 8, null)));
final SeekableStreamSupervisorSpec oldSpec =
seed.toBuilder().suspended(false).build();
final SeekableStreamSupervisorSpec newSpec =
seed.toBuilder().suspended(true).build();
- Assert.assertTrue(oldSpec.requireRestart(newSpec));
+
Assert.assertEquals(SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
oldSpec.getActionOnUpdateTo(newSpec));
}
@Test
@@ -1479,7 +1480,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
buildSpecWithIoConfig("id", createIOConfig(2,
lagBasedAutoScalerConfig(1, 8, null)));
final SeekableStreamSupervisorSpec oldSpec = seed.toBuilder().build();
final SeekableStreamSupervisorSpec newSpec =
seed.toBuilder().dataSchema(getDataSchema("other-datasource")).build();
- Assert.assertTrue(oldSpec.requireRestart(newSpec));
+
Assert.assertEquals(SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
oldSpec.getActionOnUpdateTo(newSpec));
}
@Test
@@ -1487,7 +1488,10 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
{
final TestSeekableStreamSupervisorSpec seed =
buildSpecWithIoConfig("id", createIOConfig(2,
lagBasedAutoScalerConfig(1, 8, null)));
- Assert.assertTrue(seed.toBuilder().build().requireRestart(new
NoopSupervisorSpec("id", List.of("ds"))));
+ Assert.assertEquals(
+ SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
+ seed.toBuilder().build().getActionOnUpdateTo(new
NoopSupervisorSpec("id", List.of("ds")))
+ );
}
@Test
@@ -1500,7 +1504,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
buildSpecWithIoConfig("id", createIOConfig(2,
lagBasedAutoScalerConfig(1, 8, null))).toBuilder().build();
final SeekableStreamSupervisorSpec newSpec =
buildSpecWithIoConfig("id", createIOConfig(5,
null)).toBuilder().build();
- Assert.assertTrue(oldSpec.requireRestart(newSpec));
+
Assert.assertEquals(SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
oldSpec.getActionOnUpdateTo(newSpec));
}
@Test
@@ -1511,7 +1515,7 @@ public class SeekableStreamSupervisorSpecTest extends
SeekableStreamSupervisorTe
final SeekableStreamSupervisorSpec oldSpec = seed.toBuilder().build();
final SeekableStreamSupervisorSpec newSpec =
seed.toBuilder().tuningConfig(EasyMock.mock(SeekableStreamSupervisorTuningConfig.class)).build();
- Assert.assertTrue(oldSpec.requireRestart(newSpec));
+
Assert.assertEquals(SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS,
oldSpec.getActionOnUpdateTo(newSpec));
}
private void assertMergeResult(
diff --git
a/server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpec.java
b/server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpec.java
index 95e20c7858c..6969974922a 100644
---
a/server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpec.java
+++
b/server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpec.java
@@ -141,11 +141,13 @@ public interface SupervisorSpec
}
/**
- * Returns true if replacing this (running) spec with {@code proposedSpec}
requires a supervisor restart.
- * Invoked on the running spec; default is conservative (always restart).
+ * Returns the action to take when this (running) spec is updated to {@code
proposedSpec}.
+ * Invoked on the running spec; default is conservative and always restarts
the supervisor and its tasks.
+ *
+ * @param proposedSpec the proposed supervisor spec
*/
- default boolean requireRestart(SupervisorSpec proposedSpec)
+ default SupervisorSpecUpdateAction getActionOnUpdateTo(SupervisorSpec
proposedSpec)
{
- return true;
+ return SupervisorSpecUpdateAction.RESTART_SUPERVISOR_AND_TASKS;
}
}
diff --git
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateResult.java
b/server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateAction.java
similarity index 57%
copy from
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateResult.java
copy to
server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateAction.java
index bf14917672c..45a0ab174f4 100644
---
a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateResult.java
+++
b/server/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateAction.java
@@ -17,34 +17,15 @@
* under the License.
*/
+
package org.apache.druid.indexing.overlord.supervisor;
/**
- * Outcome of {@link
SupervisorManager#createOrUpdateAndStartSupervisor(SupervisorSpec, boolean)}.
+ * Enum representing the possible actions to take when a supervisor spec is
updated.
*/
-public final class SupervisorSpecUpdateResult
+public enum SupervisorSpecUpdateAction
{
- private final boolean modified;
- private final boolean restarted;
-
- private SupervisorSpecUpdateResult(final boolean modified, final boolean
restarted)
- {
- this.modified = modified;
- this.restarted = restarted;
- }
-
- public static SupervisorSpecUpdateResult of(final boolean modified, final
boolean restarted)
- {
- return new SupervisorSpecUpdateResult(modified, restarted);
- }
-
- public boolean isModified()
- {
- return modified;
- }
-
- public boolean isRestarted()
- {
- return restarted;
- }
+ NONE,
+ RESTART_SUPERVISOR,
+ RESTART_SUPERVISOR_AND_TASKS
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
