kfaraz commented on code in PR #19541:
URL: https://github.com/apache/druid/pull/19541#discussion_r3440602955
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/LagAggregator.java:
##########
@@ -48,6 +49,28 @@ public interface LagAggregator
*/
class DefaultLagAggregator implements LagAggregator
{
+ public DefaultLagAggregator()
Review Comment:
This should be private since we want to use this class as a singleton.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -303,4 +355,87 @@ public abstract SeekableStreamSupervisorSpec
createBackfillSpec(
@Nullable Integer taskCount
);
+ /**
+ * Copy builder for restart comparison. Subclasses override; default
requires restart on any change.
+ */
+ public Builder<?> toBuilder()
Review Comment:
Please make this method abstract so that extensions do not forget to
implement this and eventually encounter exception when trying to update a spec.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java:
##########
@@ -165,29 +165,40 @@ public Response specPost(
.build();
}
- if (Boolean.TRUE.equals(skipRestartIfUnmodified) &&
!manager.shouldUpdateSupervisor(spec)) {
- return Response.ok(ImmutableMap.of("id", spec.getId(),
"restarted", false)).build();
- }
+ final SupervisorSpecUpdateResult updateResult =
+ manager.createOrUpdateAndStartSupervisor(spec,
Boolean.TRUE.equals(skipRestartIfUnmodified));
- manager.createOrUpdateAndStartSupervisor(spec);
-
- final String auditPayload
- = StringUtils.format("Update supervisor[%s] for datasource[%s]",
spec.getId(), spec.getDataSources());
- auditManager.doAudit(
- AuditEntry.builder()
- .key(spec.getId())
- .type("supervisor")
- .auditInfo(AuthorizationUtils.buildAuditInfo(req))
-
.request(AuthorizationUtils.buildRequestInfo("overlord", req))
- .payload(auditPayload)
- .build()
- );
+ if (updateResult.isModified() || updateResult.isRestarted()) {
Review Comment:
Nit: Should restart in the absence of a modification also require an audit?
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorSpec.java:
##########
@@ -295,6 +297,56 @@ public void merge(@Nullable SupervisorSpec existingSpec)
}
}
+ @Override
+ public boolean requireRestart(SupervisorSpec proposedSpec)
+ {
+ if (!(proposedSpec instanceof SeekableStreamSupervisorSpec proposed)) {
+ return true;
+ }
+
+ final Builder<?> proposedCopy;
+ try {
+ proposedCopy = proposed.toBuilder();
+ }
+ catch (UnsupportedOperationException e) {
+ return true;
+ }
+
+ if (isAutoScalerEnabled() || proposed.isAutoScalerEnabled()) {
+ proposedCopy.taskCount(getIoConfig().getTaskCount());
Review Comment:
I suppose this means that changes to `taskCount` will not trigger a restart.
Could you please add a short comment here explaining why it is okay to
ignore this?
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -186,47 +188,81 @@ public boolean
createOrUpdateAndStartSupervisor(SupervisorSpec spec)
synchronized (lock) {
Preconditions.checkState(started, "SupervisorManager not started");
- final boolean shouldUpdateSpec = shouldUpdateSupervisor(spec);
- SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
- spec.merge(existingSpec);
- createAndStartSupervisorInternal(spec, shouldUpdateSpec);
- return shouldUpdateSpec;
+
+ if (!skipRestartIfUnmodified) {
+ // Always stop/recreate, persisting whenever the spec actually changed
(or is new).
+ final boolean specChanged = isSpecChangedAndValidate(spec);
+ final SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ spec.merge(existingSpec);
+ createAndStartSupervisorInternal(spec, specChanged);
+ return SupervisorSpecUpdateResult.of(specChanged, true);
+ }
+
+ final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
+ final boolean isNew = current == null || current.rhs == null;
+
+ if (isNew) {
+ spec.merge(null);
Review Comment:
This seems unnecessary and error-prone.
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -186,47 +188,81 @@ public boolean
createOrUpdateAndStartSupervisor(SupervisorSpec spec)
synchronized (lock) {
Preconditions.checkState(started, "SupervisorManager not started");
- final boolean shouldUpdateSpec = shouldUpdateSupervisor(spec);
- SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
- spec.merge(existingSpec);
- createAndStartSupervisorInternal(spec, shouldUpdateSpec);
- return shouldUpdateSpec;
+
+ if (!skipRestartIfUnmodified) {
+ // Always stop/recreate, persisting whenever the spec actually changed
(or is new).
+ final boolean specChanged = isSpecChangedAndValidate(spec);
+ final SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ spec.merge(existingSpec);
+ createAndStartSupervisorInternal(spec, specChanged);
+ return SupervisorSpecUpdateResult.of(specChanged, true);
+ }
+
+ final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
+ final boolean isNew = current == null || current.rhs == null;
+
+ if (isNew) {
+ spec.merge(null);
+ createAndStartSupervisorInternal(spec, true);
+ return SupervisorSpecUpdateResult.of(true, true);
+ }
+
+ if (!isSpecChanged(spec, current.rhs)) {
+ return SupervisorSpecUpdateResult.of(false, false);
+ }
+
+ // merge() may carry forward omitted fields (e.g. taskCount); compare
the effective spec.
+ spec.merge(current.rhs);
+ if (!isSpecChanged(spec, current.rhs)) {
+ return SupervisorSpecUpdateResult.of(false, false);
+ }
+
+ // The effective (merged) spec is what will be persisted, so validate
that transition exactly once.
+ current.rhs.validateSpecUpdateTo(spec);
+
+ if (!current.rhs.requireRestart(spec)) {
+ metadataSupervisorManager.insert(spec.getId(), spec);
+ supervisors.put(spec.getId(), Pair.of(current.lhs, spec));
+ return SupervisorSpecUpdateResult.of(true, false);
+ }
+
+ // Restart path: stop+recreate, persisting the changed spec.
+ possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ createAndStartSupervisorInternal(spec, true);
+ return SupervisorSpecUpdateResult.of(true, true);
}
}
+ /** Byte-level diff against the running spec; validates allowed updates when
bytes differ. */
+ private boolean isSpecChangedAndValidate(SupervisorSpec spec)
+ {
+ final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
+ final SupervisorSpec currentSpec = current == null ? null : current.rhs;
+ if (!isSpecChanged(spec, currentSpec)) {
+ return false;
+ }
+ if (currentSpec != null) {
+ currentSpec.validateSpecUpdateTo(spec);
+ }
+ return true;
+ }
+
/**
- * Checks whether the submitted SupervisorSpec differs from the current spec
in SupervisorManager's supervisor list.
- * This is used in SupervisorResource specPost to determine whether the
Supervisor needs to be restarted
- *
- * @param spec The spec submitted
- * @return boolean - true only if the spec has been modified, false otherwise
+ * Byte-level diff of {@code spec} against {@code against}. A missing {@code
against} (new supervisor)
Review Comment:
How can the new supervisor be missing?
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java:
##########
@@ -165,29 +165,40 @@ public Response specPost(
.build();
}
- if (Boolean.TRUE.equals(skipRestartIfUnmodified) &&
!manager.shouldUpdateSupervisor(spec)) {
- return Response.ok(ImmutableMap.of("id", spec.getId(),
"restarted", false)).build();
- }
+ final SupervisorSpecUpdateResult updateResult =
+ manager.createOrUpdateAndStartSupervisor(spec,
Boolean.TRUE.equals(skipRestartIfUnmodified));
- manager.createOrUpdateAndStartSupervisor(spec);
-
- final String auditPayload
- = StringUtils.format("Update supervisor[%s] for datasource[%s]",
spec.getId(), spec.getDataSources());
- auditManager.doAudit(
- AuditEntry.builder()
- .key(spec.getId())
- .type("supervisor")
- .auditInfo(AuthorizationUtils.buildAuditInfo(req))
-
.request(AuthorizationUtils.buildRequestInfo("overlord", req))
- .payload(auditPayload)
- .build()
- );
+ if (updateResult.isModified() || updateResult.isRestarted()) {
+ auditSupervisorUpdate(spec, req);
+ }
- return Response.ok(ImmutableMap.of("id", spec.getId(), "restarted",
true)).build();
+ return Response.ok(
+ Map.of(
+ "id", spec.getId(),
+ "modified", updateResult.isModified(),
+ "restarted", updateResult.isRestarted()
+ )
+ ).build();
}
);
}
+ /** Audits supervisor spec submissions that changed or restarted the
supervisor. */
Review Comment:
```suggestion
/**
* Audits supervisor spec submissions that changed or restarted the
supervisor.
*/
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -186,47 +188,81 @@ public boolean
createOrUpdateAndStartSupervisor(SupervisorSpec spec)
synchronized (lock) {
Preconditions.checkState(started, "SupervisorManager not started");
- final boolean shouldUpdateSpec = shouldUpdateSupervisor(spec);
- SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
- spec.merge(existingSpec);
- createAndStartSupervisorInternal(spec, shouldUpdateSpec);
- return shouldUpdateSpec;
+
+ if (!skipRestartIfUnmodified) {
+ // Always stop/recreate, persisting whenever the spec actually changed
(or is new).
+ final boolean specChanged = isSpecChangedAndValidate(spec);
+ final SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ spec.merge(existingSpec);
+ createAndStartSupervisorInternal(spec, specChanged);
+ return SupervisorSpecUpdateResult.of(specChanged, true);
+ }
+
+ final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
+ final boolean isNew = current == null || current.rhs == null;
+
+ if (isNew) {
+ spec.merge(null);
+ createAndStartSupervisorInternal(spec, true);
+ return SupervisorSpecUpdateResult.of(true, true);
+ }
+
+ if (!isSpecChanged(spec, current.rhs)) {
+ return SupervisorSpecUpdateResult.of(false, false);
+ }
+
+ // merge() may carry forward omitted fields (e.g. taskCount); compare
the effective spec.
+ spec.merge(current.rhs);
+ if (!isSpecChanged(spec, current.rhs)) {
+ return SupervisorSpecUpdateResult.of(false, false);
+ }
+
+ // The effective (merged) spec is what will be persisted, so validate
that transition exactly once.
+ current.rhs.validateSpecUpdateTo(spec);
+
+ if (!current.rhs.requireRestart(spec)) {
+ metadataSupervisorManager.insert(spec.getId(), spec);
+ supervisors.put(spec.getId(), Pair.of(current.lhs, spec));
+ return SupervisorSpecUpdateResult.of(true, false);
+ }
+
+ // Restart path: stop+recreate, persisting the changed spec.
+ possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ createAndStartSupervisorInternal(spec, true);
+ return SupervisorSpecUpdateResult.of(true, true);
}
}
+ /** Byte-level diff against the running spec; validates allowed updates when
bytes differ. */
+ private boolean isSpecChangedAndValidate(SupervisorSpec spec)
Review Comment:
```suggestion
private boolean isSpecChangedAndValidated(SupervisorSpec spec)
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -186,47 +188,81 @@ public boolean
createOrUpdateAndStartSupervisor(SupervisorSpec spec)
synchronized (lock) {
Preconditions.checkState(started, "SupervisorManager not started");
- final boolean shouldUpdateSpec = shouldUpdateSupervisor(spec);
- SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
- spec.merge(existingSpec);
- createAndStartSupervisorInternal(spec, shouldUpdateSpec);
- return shouldUpdateSpec;
+
+ if (!skipRestartIfUnmodified) {
+ // Always stop/recreate, persisting whenever the spec actually changed
(or is new).
+ final boolean specChanged = isSpecChangedAndValidate(spec);
+ final SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ spec.merge(existingSpec);
+ createAndStartSupervisorInternal(spec, specChanged);
+ return SupervisorSpecUpdateResult.of(specChanged, true);
+ }
+
+ final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
+ final boolean isNew = current == null || current.rhs == null;
+
+ if (isNew) {
+ spec.merge(null);
+ createAndStartSupervisorInternal(spec, true);
+ return SupervisorSpecUpdateResult.of(true, true);
+ }
+
+ if (!isSpecChanged(spec, current.rhs)) {
+ return SupervisorSpecUpdateResult.of(false, false);
+ }
+
+ // merge() may carry forward omitted fields (e.g. taskCount); compare
the effective spec.
+ spec.merge(current.rhs);
+ if (!isSpecChanged(spec, current.rhs)) {
+ return SupervisorSpecUpdateResult.of(false, false);
+ }
+
+ // The effective (merged) spec is what will be persisted, so validate
that transition exactly once.
+ current.rhs.validateSpecUpdateTo(spec);
+
+ if (!current.rhs.requireRestart(spec)) {
+ metadataSupervisorManager.insert(spec.getId(), spec);
+ supervisors.put(spec.getId(), Pair.of(current.lhs, spec));
+ return SupervisorSpecUpdateResult.of(true, false);
+ }
+
+ // Restart path: stop+recreate, persisting the changed spec.
+ possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ createAndStartSupervisorInternal(spec, true);
+ return SupervisorSpecUpdateResult.of(true, true);
}
}
+ /** Byte-level diff against the running spec; validates allowed updates when
bytes differ. */
+ private boolean isSpecChangedAndValidate(SupervisorSpec spec)
+ {
+ final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
+ final SupervisorSpec currentSpec = current == null ? null : current.rhs;
+ if (!isSpecChanged(spec, currentSpec)) {
+ return false;
+ }
+ if (currentSpec != null) {
+ currentSpec.validateSpecUpdateTo(spec);
+ }
+ return true;
+ }
+
/**
- * Checks whether the submitted SupervisorSpec differs from the current spec
in SupervisorManager's supervisor list.
- * This is used in SupervisorResource specPost to determine whether the
Supervisor needs to be restarted
- *
- * @param spec The spec submitted
- * @return boolean - true only if the spec has been modified, false otherwise
+ * Byte-level diff of {@code spec} against {@code against}. A missing {@code
against} (new supervisor)
+ * counts as changed. Does not validate the transition.
*/
- public boolean shouldUpdateSupervisor(SupervisorSpec spec)
+ private boolean isSpecChanged(SupervisorSpec spec, @Nullable SupervisorSpec
against)
Review Comment:
Nit: Maybe use the arg names `currentSpec` and `proposedSpec` for clarity.
```suggestion
private boolean areSpecBytesChanged(SupervisorSpec proposedSpec, @Nullable
SupervisorSpec currentSpec)
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -186,47 +188,81 @@ public boolean
createOrUpdateAndStartSupervisor(SupervisorSpec spec)
synchronized (lock) {
Preconditions.checkState(started, "SupervisorManager not started");
- final boolean shouldUpdateSpec = shouldUpdateSupervisor(spec);
- SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
- spec.merge(existingSpec);
- createAndStartSupervisorInternal(spec, shouldUpdateSpec);
- return shouldUpdateSpec;
+
+ if (!skipRestartIfUnmodified) {
+ // Always stop/recreate, persisting whenever the spec actually changed
(or is new).
+ final boolean specChanged = isSpecChangedAndValidate(spec);
+ final SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ spec.merge(existingSpec);
+ createAndStartSupervisorInternal(spec, specChanged);
+ return SupervisorSpecUpdateResult.of(specChanged, true);
+ }
+
+ final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
+ final boolean isNew = current == null || current.rhs == null;
+
+ if (isNew) {
+ spec.merge(null);
+ createAndStartSupervisorInternal(spec, true);
+ return SupervisorSpecUpdateResult.of(true, true);
+ }
+
+ if (!isSpecChanged(spec, current.rhs)) {
+ return SupervisorSpecUpdateResult.of(false, false);
+ }
+
+ // merge() may carry forward omitted fields (e.g. taskCount); compare
the effective spec.
+ spec.merge(current.rhs);
+ if (!isSpecChanged(spec, current.rhs)) {
+ return SupervisorSpecUpdateResult.of(false, false);
+ }
+
+ // The effective (merged) spec is what will be persisted, so validate
that transition exactly once.
+ current.rhs.validateSpecUpdateTo(spec);
+
+ if (!current.rhs.requireRestart(spec)) {
+ metadataSupervisorManager.insert(spec.getId(), spec);
+ supervisors.put(spec.getId(), Pair.of(current.lhs, spec));
+ return SupervisorSpecUpdateResult.of(true, false);
+ }
+
+ // Restart path: stop+recreate, persisting the changed spec.
+ possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ createAndStartSupervisorInternal(spec, true);
+ return SupervisorSpecUpdateResult.of(true, true);
}
}
+ /** Byte-level diff against the running spec; validates allowed updates when
bytes differ. */
+ private boolean isSpecChangedAndValidate(SupervisorSpec spec)
+ {
+ final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
+ final SupervisorSpec currentSpec = current == null ? null : current.rhs;
+ if (!isSpecChanged(spec, currentSpec)) {
+ return false;
+ }
+ if (currentSpec != null) {
+ currentSpec.validateSpecUpdateTo(spec);
+ }
+ return true;
Review Comment:
Please use a simpler if-else chain to clarify the flow here.
```suggestion
if (currentSpec == null) {
return true;
} else if (!isSpecChanged(spec, currentSpec)) {
return false;
} else {
return currentSpec.validateSpecUpdateTo(spec);
}
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java:
##########
@@ -186,47 +188,81 @@ public boolean
createOrUpdateAndStartSupervisor(SupervisorSpec spec)
synchronized (lock) {
Preconditions.checkState(started, "SupervisorManager not started");
- final boolean shouldUpdateSpec = shouldUpdateSupervisor(spec);
- SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
- spec.merge(existingSpec);
- createAndStartSupervisorInternal(spec, shouldUpdateSpec);
- return shouldUpdateSpec;
+
+ if (!skipRestartIfUnmodified) {
+ // Always stop/recreate, persisting whenever the spec actually changed
(or is new).
+ final boolean specChanged = isSpecChangedAndValidate(spec);
+ final SupervisorSpec existingSpec =
possiblyStopAndRemoveSupervisorInternal(spec.getId(), false);
+ spec.merge(existingSpec);
+ createAndStartSupervisorInternal(spec, specChanged);
+ return SupervisorSpecUpdateResult.of(specChanged, true);
+ }
+
+ final Pair<Supervisor, SupervisorSpec> current =
supervisors.get(spec.getId());
+ final boolean isNew = current == null || current.rhs == null;
+
+ if (isNew) {
+ spec.merge(null);
+ createAndStartSupervisorInternal(spec, true);
+ return SupervisorSpecUpdateResult.of(true, true);
+ }
+
+ if (!isSpecChanged(spec, current.rhs)) {
Review Comment:
Assign currentSpec to a variable for readability:
```suggestion
final SupervisorSpec currentSpec = current.rhs;
if (!isSpecChanged(spec, currentSpec)) {
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
