You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by kf...@apache.org on 2026/07/15 07:43:02 UTC
(druid) branch master updated: refactor: make supervisor diff+restart algorithm smarter on submission (#19541)
This is an automated email from the ASF dual-hosted git repository.
kfaraz 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 355a9b83d40 refactor: make supervisor diff+restart algorithm smarter on submission (#19541)
355a9b83d40 is described below
commit 355a9b83d403c3df875fc3a548351f9f79c6d9e2
Author: jtuglu1 <jt...@netflix.com>
AuthorDate: Wed Jul 15 00:42:53 2026 -0700
refactor: make supervisor diff+restart algorithm smarter on submission (#19541)
Restarting a supervisor spec after submission sometimes isn't always needed and
can be expensive for large supervisors. The current skip restart logic is very bare-bones
(doing effectively a byte-level check between the 2 specs) instead of allowing each
spec to override its own restart requirements, often causing some false positive restarts.
This diff can also be affected by false-positive diffs like ioConfig.taskCount being updated
while auto-scaling is configured, whitespace changes, etc.
Changes:
- Add `SupervisorSpec.requireRestart()` to check if a restart is actually required
- Update response of `SupervisorResource.specPost()` to include a `restarted` field along with `modified`
- Add comparators and spec builders to simplify comparison of specs during an update
- No change to spec persistence behavior i.e. byte-level differing spec updates are persisted into the DB,
but may not necessarily invoke a supervisor restart.
---
.../IngestionBackwardCompatibilityDockerTest.java | 8 +
.../embedded/indexing/IngestionSmokeTest.java | 45 ++-
.../RabbitStreamIndexTaskTuningConfig.java | 25 ++
.../supervisor/RabbitStreamIOConfigBuilder.java | 69 ++--
.../supervisor/RabbitStreamSupervisorIOConfig.java | 28 ++
.../supervisor/RabbitStreamSupervisorSpec.java | 38 +++
.../RabbitStreamSupervisorTuningConfig.java | 24 ++
.../RabbitStreamIndexTaskTuningConfigTest.java | 17 +
.../RabbitStreamSupervisorIOConfigTest.java | 72 ++++
.../supervisor/RabbitStreamSupervisorTest.java | 139 +++-----
.../RabbitStreamSupervisorTuningConfigTest.java | 74 ++++
.../kafka/supervisor/KafkaIOConfigBuilder.java | 45 ++-
.../kafka/supervisor/KafkaSupervisorIOConfig.java | 39 +++
.../kafka/supervisor/KafkaSupervisorSpec.java | 38 +++
.../supervisor/KafkaSupervisorTuningConfig.java | 24 ++
.../supervisor/KafkaSupervisorIOConfigTest.java | 83 +++++
.../kafka/supervisor/KafkaSupervisorTest.java | 240 +++++--------
.../kinesis/supervisor/KinesisIOConfigBuilder.java | 125 +++++++
.../supervisor/KinesisSupervisorIOConfig.java | 39 +++
.../kinesis/supervisor/KinesisSupervisorSpec.java | 47 +++
.../supervisor/KinesisSupervisorTuningConfig.java | 35 ++
.../indexing/kinesis/KinesisSamplerSpecTest.java | 61 +---
.../supervisor/KinesisSupervisorIOConfigTest.java | 73 ++++
.../kinesis/supervisor/KinesisSupervisorTest.java | 371 ++++++++-------------
.../supervisor/CompactionSupervisorManager.java | 5 +-
.../overlord/supervisor/SupervisorManager.java | 120 ++++---
.../overlord/supervisor/SupervisorResource.java | 45 ++-
.../supervisor/SupervisorSpecUpdateResult.java | 50 +++
.../seekablestream/supervisor/IdleConfig.java | 20 ++
.../seekablestream/supervisor/LagAggregator.java | 23 ++
.../SeekableStreamSupervisorIOConfig.java | 59 ++++
.../SeekableStreamSupervisorIngestionSpec.java | 23 ++
.../supervisor/SeekableStreamSupervisorSpec.java | 130 ++++++++
.../supervisor/SupervisorIOConfigBuilder.java | 99 +++++-
.../SegmentTransactionalInsertActionTest.java | 2 +-
.../http/OverlordCompactionResourceTest.java | 7 +-
.../overlord/supervisor/SupervisorManagerTest.java | 275 +++++++++++++--
.../supervisor/SupervisorResourceTest.java | 69 +++-
.../SeekableStreamSamplerSpecTest.java | 7 +
.../SeekableStreamSupervisorIOConfigTest.java | 147 +++++++-
.../SeekableStreamSupervisorSpecTest.java | 350 ++++++++++---------
.../SeekableStreamSupervisorStateTest.java | 280 ++++++----------
.../SeekableStreamSupervisorTestBase.java | 84 +++--
.../overlord/supervisor/SupervisorSpec.java | 9 +
44 files changed, 2508 insertions(+), 1055 deletions(-)
diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/docker/IngestionBackwardCompatibilityDockerTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/docker/IngestionBackwardCompatibilityDockerTest.java
index ad3855efacd..00912129fd2 100644
--- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/docker/IngestionBackwardCompatibilityDockerTest.java
+++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/docker/IngestionBackwardCompatibilityDockerTest.java
@@ -36,6 +36,7 @@ import org.apache.druid.testing.embedded.indexing.IngestionSmokeTest;
import org.jboss.netty.handler.codec.http.HttpMethod;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
import java.util.Map;
@@ -120,6 +121,13 @@ public class IngestionBackwardCompatibilityDockerTest extends IngestionSmokeTest
Assertions.assertEquals(Map.of("id", supervisorId), startSupervisorResult);
}
+ @Override
+ @Disabled("modified/restarted response semantics and skipRestartIfUnmodified are not supported by the old Overlord")
+ public void test_kafkaSupervisor_modifiedAndRestartedCombinations()
+ {
+ // No-op: the older Overlord returns only {id} and always restarts on submission.
+ }
+
@Override
protected void waitForNextCoordinatorCacheSync()
{
diff --git a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
index 934f67b0dda..e9033a5375a 100644
--- a/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
+++ b/embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java
@@ -59,6 +59,7 @@ import org.apache.druid.timeline.DataSegment;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.joda.time.DateTime;
import org.joda.time.Interval;
+import org.joda.time.Period;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
@@ -325,6 +326,37 @@ public class IngestionSmokeTest extends EmbeddedClusterTestBase
Assertions.assertTrue(supervisorStatus.isSuspended());
}
+ @Test
+ public void test_kafkaSupervisor_modifiedAndRestartedCombinations()
+ {
+ final String topic = dataSource;
+ kafkaServer.createTopicWithPartitions(topic, 2);
+
+ final String supervisorId = dataSource;
+
+ // (1) First submission of a new supervisor: persisted and started.
+ Assertions.assertEquals(
+ Map.of("id", supervisorId, "modified", "true", "restarted", "true"),
+ postKafkaSupervisor(createKafkaSupervisor(topic, Period.millis(500)))
+ );
+
+ // (2) Resubmitting a byte-identical spec is a no-op (the client always sets skipRestartIfUnmodified).
+ Assertions.assertEquals(
+ Map.of("id", supervisorId, "modified", "false", "restarted", "false"),
+ postKafkaSupervisor(createKafkaSupervisor(topic, Period.millis(500)))
+ );
+
+ // (3) A restart-requiring change (taskDuration): persisted and restarted.
+ Assertions.assertEquals(
+ Map.of("id", supervisorId, "modified", "true", "restarted", "true"),
+ postKafkaSupervisor(createKafkaSupervisor(topic, Period.seconds(10)))
+ );
+
+ // The (modified=true, restarted=false) case needs the autoscaler to mutate taskCount at runtime
+ // (a plain resubmit is reset to taskCountStart by merge()), so it is covered deterministically in
+ // SupervisorManagerTest#testCreateOrUpdateSkipRestart_changedNoRestart_persistsWithoutRestart.
+ }
+
@Test
public void test_streamLogs_ofCancelledTask() throws Exception
{
@@ -368,6 +400,11 @@ public class IngestionSmokeTest extends EmbeddedClusterTestBase
}
private KafkaSupervisorSpec createKafkaSupervisor(String topic)
+ {
+ return createKafkaSupervisor(topic, Period.millis(500));
+ }
+
+ private KafkaSupervisorSpec createKafkaSupervisor(String topic, Period taskDuration)
{
return MoreResources.Supervisor.KAFKA_JSON
.get()
@@ -376,11 +413,17 @@ public class IngestionSmokeTest extends EmbeddedClusterTestBase
ioConfig -> ioConfig
.withConsumerProperties(kafkaServer.consumerProperties())
.withInputFormat(new CsvInputFormat(List.of("timestamp", "item"), null, null, false, 0, false))
+ .withTaskDuration(taskDuration)
)
.withTuningConfig(tuningConfig -> tuningConfig.withMaxRowsPerSegment(1))
.build(dataSource, topic);
}
+ private Map<String, String> postKafkaSupervisor(KafkaSupervisorSpec spec)
+ {
+ return cluster.callApi().onLeaderOverlord(o -> o.postSupervisor(spec));
+ }
+
private List<ProducerRecord<byte[], byte[]>> generateRecordsForTopic(
String topic,
int numRecords,
@@ -424,7 +467,7 @@ public class IngestionSmokeTest extends EmbeddedClusterTestBase
protected void validateSupervisorUpdateResponse(Map<String, String> startSupervisorResult, String supervisorId)
{
- Assertions.assertEquals(Map.of("id", supervisorId, "restarted", "true"), startSupervisorResult);
+ Assertions.assertEquals(Map.of("id", supervisorId, "modified", "true", "restarted", "true"), startSupervisorResult);
}
protected void waitForNextCoordinatorCacheSync()
diff --git a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfig.java b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfig.java
index dd5bc84d523..24bdea6816b 100644
--- a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfig.java
+++ b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfig.java
@@ -32,6 +32,7 @@ import org.joda.time.Period;
import javax.annotation.Nullable;
import java.io.File;
+import java.util.Objects;
public class RabbitStreamIndexTaskTuningConfig extends SeekableStreamIndexTaskTuningConfig
{
@@ -238,6 +239,30 @@ public class RabbitStreamIndexTaskTuningConfig extends SeekableStreamIndexTaskTu
);
}
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ RabbitStreamIndexTaskTuningConfig that = (RabbitStreamIndexTaskTuningConfig) o;
+ return recordBufferOfferTimeout == that.recordBufferOfferTimeout
+ && Objects.equals(recordBufferSize, that.recordBufferSize)
+ && Objects.equals(maxRecordsPerPoll, that.maxRecordsPerPoll);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(super.hashCode(), recordBufferSize, recordBufferOfferTimeout, maxRecordsPerPoll);
+ }
+
@Override
public String toString()
{
diff --git a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaIOConfigBuilder.java b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamIOConfigBuilder.java
similarity index 52%
copy from extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaIOConfigBuilder.java
copy to extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamIOConfigBuilder.java
index ac8a9e1d2f7..729ff7c9097 100644
--- a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaIOConfigBuilder.java
+++ b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamIOConfigBuilder.java
@@ -17,91 +17,74 @@
* under the License.
*/
-package org.apache.druid.indexing.kafka.supervisor;
+package org.apache.druid.indexing.rabbitstream.supervisor;
-import org.apache.druid.data.input.InputFormat;
-import org.apache.druid.data.input.kafkainput.KafkaInputFormat;
-import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import org.apache.druid.indexing.seekablestream.supervisor.SupervisorIOConfigBuilder;
import java.util.Map;
/**
- * Builder for {@link KafkaSupervisorIOConfig}.
+ * Builder for {@link RabbitStreamSupervisorIOConfig}.
*/
-public class KafkaIOConfigBuilder extends SupervisorIOConfigBuilder<KafkaIOConfigBuilder, KafkaSupervisorIOConfig>
+public class RabbitStreamIOConfigBuilder
+ extends SupervisorIOConfigBuilder<RabbitStreamIOConfigBuilder, RabbitStreamSupervisorIOConfig>
{
- private String topic;
- private String topicPattern;
+ private String uri;
private Map<String, Object> consumerProperties;
- private BoundedStreamConfig boundedStreamConfig;
+ private Long pollTimeout;
- public KafkaIOConfigBuilder withTopic(String topic)
+ public RabbitStreamIOConfigBuilder withUri(String uri)
{
- this.topic = topic;
+ this.uri = uri;
return this;
}
- public KafkaIOConfigBuilder withTopicPattern(String topicPattern)
- {
- this.topicPattern = topicPattern;
- return this;
- }
-
- public KafkaIOConfigBuilder withConsumerProperties(Map<String, Object> consumerProperties)
+ public RabbitStreamIOConfigBuilder withConsumerProperties(Map<String, Object> consumerProperties)
{
this.consumerProperties = consumerProperties;
return this;
}
- public KafkaIOConfigBuilder withKafkaInputFormat(InputFormat valueFormat)
+ public RabbitStreamIOConfigBuilder withPollTimeout(Long pollTimeout)
{
- this.inputFormat = new KafkaInputFormat(
- null,
- null,
- valueFormat,
- null,
- null,
- null,
- null,
- null,
- null
- );
+ this.pollTimeout = pollTimeout;
return this;
}
- public KafkaIOConfigBuilder withBoundedStreamConfig(BoundedStreamConfig boundedStreamConfig)
+ /**
+ * Populates this builder (base and Rabbit-specific fields) from an existing config.
+ */
+ public RabbitStreamIOConfigBuilder copyFrom(RabbitStreamSupervisorIOConfig io)
{
- this.boundedStreamConfig = boundedStreamConfig;
+ copyFromBase(io);
+ this.uri = io.getUri();
+ this.consumerProperties = io.getConsumerProperties();
+ this.pollTimeout = io.getPollTimeout();
return this;
}
@Override
- public KafkaSupervisorIOConfig build()
+ public RabbitStreamSupervisorIOConfig build()
{
- return new KafkaSupervisorIOConfig(
- topic,
- topicPattern,
+ return new RabbitStreamSupervisorIOConfig(
+ stream,
+ uri,
inputFormat,
replicas,
taskCount,
taskDuration,
consumerProperties,
autoScalerConfig,
- lagAggregator,
- null,
+ pollTimeout,
startDelay,
period,
- useEarliestSequenceNumber,
completionTimeout,
+ useEarliestSequenceNumber,
lateMessageRejectionPeriod,
earlyMessageRejectionPeriod,
lateMessageRejectionStartDateTime,
- null,
- idleConfig,
stopTaskCount,
- null,
- null,
+ serverPriorityToReplicas,
boundedStreamConfig
);
}
diff --git a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorIOConfig.java b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorIOConfig.java
index 20c4e92988b..dd85d1dcfeb 100644
--- a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorIOConfig.java
+++ b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorIOConfig.java
@@ -33,6 +33,7 @@ import org.joda.time.Period;
import javax.annotation.Nullable;
import java.util.Map;
+import java.util.Objects;
public class RabbitStreamSupervisorIOConfig extends SeekableStreamSupervisorIOConfig
{
@@ -145,4 +146,31 @@ public class RabbitStreamSupervisorIOConfig extends SeekableStreamSupervisorIOCo
'}';
}
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ RabbitStreamSupervisorIOConfig that = (RabbitStreamSupervisorIOConfig) o;
+ return pollTimeout == that.pollTimeout
+ && Objects.equals(uri, that.uri)
+ && Objects.equals(consumerProperties, that.consumerProperties);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(super.hashCode(), uri, consumerProperties, pollTimeout);
+ }
+
+ @Override
+ public RabbitStreamIOConfigBuilder toBuilder()
+ {
+ return new RabbitStreamIOConfigBuilder().copyFrom(this);
+ }
+
}
diff --git a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorSpec.java b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorSpec.java
index 4763a949a61..c92e3f5a553 100644
--- a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorSpec.java
+++ b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorSpec.java
@@ -216,4 +216,42 @@ public class RabbitStreamSupervisorSpec extends SeekableStreamSupervisorSpec
", suspend=" + isSuspended() +
'}';
}
+
+ @Override
+ public Builder toBuilder()
+ {
+ return new Builder().copyFrom(this);
+ }
+
+ public static class Builder extends SeekableStreamSupervisorSpec.Builder<Builder>
+ {
+ @Override
+ protected Builder self()
+ {
+ return this;
+ }
+
+ @Override
+ public RabbitStreamSupervisorSpec build()
+ {
+ return new RabbitStreamSupervisorSpec(
+ id,
+ null,
+ dataSchema,
+ (RabbitStreamSupervisorTuningConfig) tuningConfig,
+ (RabbitStreamSupervisorIOConfig) ioConfig,
+ context,
+ suspended,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null
+ );
+ }
+ }
}
diff --git a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTuningConfig.java b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTuningConfig.java
index a2667026fff..28a4f55f0d7 100644
--- a/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTuningConfig.java
+++ b/extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTuningConfig.java
@@ -29,6 +29,7 @@ import org.joda.time.Duration;
import org.joda.time.Period;
import javax.annotation.Nullable;
+import java.util.Objects;
public class RabbitStreamSupervisorTuningConfig extends RabbitStreamIndexTaskTuningConfig
implements SeekableStreamSupervisorTuningConfig
@@ -185,6 +186,29 @@ public class RabbitStreamSupervisorTuningConfig extends RabbitStreamIndexTaskTun
return offsetFetchPeriod;
}
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ RabbitStreamSupervisorTuningConfig that = (RabbitStreamSupervisorTuningConfig) o;
+ return Objects.equals(workerThreads, that.workerThreads)
+ && Objects.equals(chatRetries, that.chatRetries)
+ && Objects.equals(httpTimeout, that.httpTimeout)
+ && Objects.equals(shutdownTimeout, that.shutdownTimeout)
+ && Objects.equals(offsetFetchPeriod, that.offsetFetchPeriod);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(super.hashCode(), workerThreads, chatRetries, httpTimeout, shutdownTimeout, offsetFetchPeriod);
+ }
+
@Override
public String toString()
{
diff --git a/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfigTest.java b/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfigTest.java
index 19930d1f9d5..c054e80357a 100644
--- a/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfigTest.java
+++ b/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfigTest.java
@@ -21,6 +21,8 @@ package org.apache.druid.indexing.rabbitstream;
import com.fasterxml.jackson.databind.Module;
import com.fasterxml.jackson.databind.ObjectMapper;
+import nl.jqno.equalsverifier.EqualsVerifier;
+import nl.jqno.equalsverifier.Warning;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
import org.apache.druid.segment.indexing.TuningConfig;
@@ -189,4 +191,19 @@ public class RabbitStreamIndexTaskTuningConfigTest
Assert.assertEquals(resStr, config.toString());
}
+
+ /**
+ * Drift guard for the fields this class adds ({@code recordBufferSize}, {@code recordBufferOfferTimeout},
+ * {@code maxRecordsPerPoll}): if one were dropped from {@code equals}, a tuning change would persist
+ * without restarting the supervisor.
+ */
+ @Test
+ public void testEqualsContractCoversAllFields()
+ {
+ EqualsVerifier.forClass(RabbitStreamIndexTaskTuningConfig.class)
+ .usingGetClass()
+ .withRedefinedSuperclass()
+ .suppress(Warning.NONFINAL_FIELDS)
+ .verify();
+ }
}
diff --git a/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorIOConfigTest.java b/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorIOConfigTest.java
index 0dca7f084f0..0b48d239703 100644
--- a/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorIOConfigTest.java
+++ b/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorIOConfigTest.java
@@ -22,15 +22,24 @@ package org.apache.druid.indexing.rabbitstream.supervisor;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.Module;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
+import nl.jqno.equalsverifier.EqualsVerifier;
+import nl.jqno.equalsverifier.Warning;
+import org.apache.druid.data.input.InputFormat;
import org.apache.druid.indexing.rabbitstream.RabbitStreamIndexTaskModule;
+import org.apache.druid.indexing.seekablestream.supervisor.LagAggregator;
+import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.AutoScalerConfig;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.hamcrest.CoreMatchers;
import org.joda.time.Duration;
+import org.joda.time.Period;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
+import static org.easymock.EasyMock.createMock;
+
public class RabbitStreamSupervisorIOConfigTest
{
private final ObjectMapper mapper;
@@ -212,4 +221,67 @@ public class RabbitStreamSupervisorIOConfigTest
Assert.assertNull(config.getBoundedStreamConfig());
}
+ private static RabbitStreamIOConfigBuilder ioConfigBuilder()
+ {
+ return new RabbitStreamIOConfigBuilder()
+ .withStream("stream")
+ .withUri("rabbit://localhost")
+ .withReplicas(1)
+ .withTaskCount(2)
+ .withTaskDuration(new Period("PT1H"));
+ }
+
+ @Test
+ public void testEqualsAndHashCode()
+ {
+ final RabbitStreamSupervisorIOConfig config = ioConfigBuilder().build();
+ Assert.assertEquals(config, ioConfigBuilder().build());
+ Assert.assertEquals(config.hashCode(), ioConfigBuilder().build().hashCode());
+ Assert.assertNotEquals(config, null);
+ Assert.assertNotEquals(config, "not an io config");
+ Assert.assertNotEquals(config, ioConfigBuilder().withUri("rabbit://other").build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withReplicas(9).build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withTaskCount(9).build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withPollTimeout(999L).build());
+ }
+
+ @Test
+ public void testTuningConfigEqualsAndHashCode()
+ {
+ final RabbitStreamSupervisorTuningConfig config = RabbitStreamSupervisorTuningConfig.defaultConfig();
+ Assert.assertEquals(config, RabbitStreamSupervisorTuningConfig.defaultConfig());
+ Assert.assertEquals(config.hashCode(), RabbitStreamSupervisorTuningConfig.defaultConfig().hashCode());
+ Assert.assertNotEquals(config, null);
+ Assert.assertNotEquals(config, "not a tuning config");
+ }
+
+ /**
+ * Drift guard for this class's own fields (base fields are covered by SeekableStreamSupervisorIOConfigTest):
+ * a field omitted from {@code equals} would let a changed spec persist without restarting the supervisor.
+ */
+ @Test
+ public void testEqualsContractCoversAllFields()
+ {
+ EqualsVerifier.forClass(RabbitStreamSupervisorIOConfig.class)
+ .usingGetClass()
+ .withRedefinedSuperclass()
+ .withIgnoredFields("taskCountExplicit", "autoScalerEnabled")
+ .suppress(Warning.NONFINAL_FIELDS)
+ .withPrefabValues(Optional.class, Optional.of("a"), Optional.of("b"))
+ .withPrefabValues(InputFormat.class, createMock(InputFormat.class), createMock(InputFormat.class))
+ .withPrefabValues(AutoScalerConfig.class, createMock(AutoScalerConfig.class), createMock(AutoScalerConfig.class))
+ .withPrefabValues(LagAggregator.class, createMock(LagAggregator.class), createMock(LagAggregator.class))
+ .verify();
+ }
+
+ @Test
+ public void testTuningConfigEqualsContractCoversAllFields()
+ {
+ EqualsVerifier.forClass(RabbitStreamSupervisorTuningConfig.class)
+ .usingGetClass()
+ .withRedefinedSuperclass()
+ .suppress(Warning.NONFINAL_FIELDS)
+ .verify();
+ }
+
}
diff --git a/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTest.java b/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTest.java
index 82ee7afd0e0..0387c8c7517 100644
--- a/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTest.java
+++ b/extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorTest.java
@@ -196,27 +196,22 @@ public class RabbitStreamSupervisorTest extends EasyMockSupport
DataSchema dataSchema,
RabbitStreamSupervisorTuningConfig tuningConfig)
{
- RabbitStreamSupervisorIOConfig rabbitStreamSupervisorIOConfig = new RabbitStreamSupervisorIOConfig(
- STREAM, // stream
- URI, // uri
- INPUT_FORMAT, // inputFormat
- replicas, // replicas
- taskCount, // taskCount
- new Period(duration), // taskDuration
- null, // consumerProperties
- null, // autoscalerConfig
- 400L, // poll timeout
- new Period("P1D"), // start delat
- new Period("PT30M"), // period
- new Period("PT30S"), // completiontimeout
- false, // useearliest
- lateMessageRejectionPeriod, // latemessagerejection
- earlyMessageRejectionPeriod, // early message rejection
- null, // latemessagerejectionstartdatetime
- 1,
- null,
- null
- );
+ RabbitStreamSupervisorIOConfig rabbitStreamSupervisorIOConfig = new RabbitStreamIOConfigBuilder()
+ .withStream(STREAM)
+ .withUri(URI)
+ .withInputFormat(INPUT_FORMAT)
+ .withReplicas(replicas)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period(duration))
+ .withPollTimeout(400L)
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30M"))
+ .withCompletionTimeout(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withLateMessageRejectionPeriod(lateMessageRejectionPeriod)
+ .withEarlyMessageRejectionPeriod(earlyMessageRejectionPeriod)
+ .withStopTaskCount(1)
+ .build();
RabbitStreamIndexTaskClientFactory clientFactory = new RabbitStreamIndexTaskClientFactory(null,
OBJECT_MAPPER);
RabbitStreamSupervisor supervisor = new RabbitStreamSupervisor(
@@ -263,27 +258,20 @@ public class RabbitStreamSupervisorTest extends EasyMockSupport
@Test
public void testRecordSupplier()
{
- RabbitStreamSupervisorIOConfig rabbitStreamSupervisorIOConfig = new RabbitStreamSupervisorIOConfig(
- STREAM, // stream
- URI, // uri
- INPUT_FORMAT, // inputFormat
- 1, // replicas
- 1, // taskCount
- new Period("PT30M"), // taskDuration
- null, // consumerProperties
- null, // autoscalerConfig
- 400L, // poll timeout
- new Period("P1D"), // start delat
- new Period("PT30M"), // period
- new Period("PT30S"), // completiontimeout
- false, // useearliest
- null, // latemessagerejection
- null, // early message rejection
- null, // latemessagerejectionstartdatetime
- 1,
- null,
- null
- );
+ RabbitStreamSupervisorIOConfig rabbitStreamSupervisorIOConfig = new RabbitStreamIOConfigBuilder()
+ .withStream(STREAM)
+ .withUri(URI)
+ .withInputFormat(INPUT_FORMAT)
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT30M"))
+ .withPollTimeout(400L)
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30M"))
+ .withCompletionTimeout(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withStopTaskCount(1)
+ .build();
RabbitStreamIndexTaskClientFactory clientFactory = new RabbitStreamIndexTaskClientFactory(null,
OBJECT_MAPPER);
RabbitStreamSupervisor supervisor = new RabbitStreamSupervisor(
@@ -407,27 +395,20 @@ public class RabbitStreamSupervisorTest extends EasyMockSupport
null,
null,
ImmutableSet.of(),
- new RabbitStreamSupervisorIOConfig(
- STREAM, // stream
- URI, // uri
- INPUT_FORMAT, // inputFormat
- 1, // replicas
- 1, // taskCount
- new Period("PT30M"), // taskDuration
- null, // consumerProperties
- null, // autoscalerConfig
- 400L, // poll timeout
- new Period("P1D"), // start delat
- new Period("PT30M"), // period
- new Period("PT30S"), // completiontimeout
- false, // useearliest
- null, // latemessagerejection
- null, // early message rejection
- null, // latemessagerejectionstartdatetime
- 1,
- null,
- null
- )
+ new RabbitStreamIOConfigBuilder()
+ .withStream(STREAM)
+ .withUri(URI)
+ .withInputFormat(INPUT_FORMAT)
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT30M"))
+ .withPollTimeout(400L)
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30M"))
+ .withCompletionTimeout(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withStopTaskCount(1)
+ .build()
);
Assert.assertEquals(30L, ioConfig.getRefreshRejectionPeriodsInMinutes().longValue());
@@ -479,27 +460,17 @@ public class RabbitStreamSupervisorTest extends EasyMockSupport
"queue-1", 600
);
- final RabbitStreamSupervisorIOConfig rabbitSupervisorIOConfig = new RabbitStreamSupervisorIOConfig(
- STREAM,
- URI,
- INPUT_FORMAT,
- 1,
- 1,
- new Period("PT30S"),
- null,
- null,
- null,
- new Period("PT30M"),
- null,
- null,
- null,
- null,
- null,
- null,
- 1000,
- null,
- new BoundedStreamConfig(startOffsets, endOffsets)
- );
+ final RabbitStreamSupervisorIOConfig rabbitSupervisorIOConfig = new RabbitStreamIOConfigBuilder()
+ .withStream(STREAM)
+ .withUri(URI)
+ .withInputFormat(INPUT_FORMAT)
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT30S"))
+ .withStartDelay(new Period("PT30M"))
+ .withStopTaskCount(1000)
+ .withBoundedStreamConfig(new BoundedStreamConfig(startOffsets, endOffsets))
+ .build();
Assert.assertTrue(rabbitSupervisorIOConfig.isBounded());
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 697306b175f..f4846573d3e 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
@@ -21,9 +21,15 @@ package org.apache.druid.indexing.rabbitstream.supervisor;
import com.fasterxml.jackson.databind.Module;
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.rabbitstream.RabbitStreamIndexTaskModule;
import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
import org.apache.druid.segment.incremental.OnheapIncrementalIndex;
+import org.apache.druid.segment.indexing.DataSchema;
import org.apache.druid.segment.indexing.TuningConfig;
import org.joda.time.Duration;
import org.joda.time.Period;
@@ -32,6 +38,9 @@ import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
+import java.util.List;
+import java.util.Map;
+
public class RabbitStreamSupervisorTuningConfigTest
{
private final ObjectMapper mapper;
@@ -45,6 +54,17 @@ public class RabbitStreamSupervisorTuningConfigTest
@Rule
public final ExpectedException exception = ExpectedException.none();
+ @Test
+ public void testRequireRestartWhenRabbitTaskTuningChanges()
+ {
+ 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))));
+ }
+
@Test
public void testSerdeWithDefaults() throws Exception
{
@@ -122,4 +142,58 @@ public class RabbitStreamSupervisorTuningConfigTest
Assert.assertEquals(Duration.standardSeconds(120), config.getRepartitionTransitionDuration());
}
+ private RabbitStreamSupervisorSpec supervisorSpec(final RabbitStreamSupervisorTuningConfig tuningConfig)
+ {
+ return new RabbitStreamSupervisorSpec.Builder()
+ .id("id")
+ .dataSchema(dataSchema())
+ .ioConfig(ioConfig())
+ .tuningConfig(tuningConfig)
+ .build();
+ }
+
+ private DataSchema dataSchema()
+ {
+ return DataSchema.builder()
+ .withDataSource("testDS")
+ .withTimestamp(new TimestampSpec("timestamp", "iso", null))
+ .withDimensions(DimensionsSpec.EMPTY)
+ .withAggregators(new CountAggregatorFactory("rows"))
+ .withGranularity(
+ new UniformGranularitySpec(
+ Granularities.HOUR,
+ Granularities.NONE,
+ List.of()
+ )
+ )
+ .build();
+ }
+
+ private RabbitStreamSupervisorIOConfig ioConfig()
+ {
+ return new RabbitStreamIOConfigBuilder()
+ .withStream("stream")
+ .withUri("rabbit://localhost")
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .build();
+ }
+
+ private RabbitStreamSupervisorTuningConfig tuningConfig(
+ final Integer recordBufferSize,
+ final Integer recordBufferOfferTimeout,
+ final Integer maxRecordsPerPoll
+ )
+ {
+ return mapper.convertValue(
+ Map.of(
+ "type", "rabbit",
+ "recordBufferSize", recordBufferSize,
+ "recordBufferOfferTimeout", recordBufferOfferTimeout,
+ "maxRecordsPerPoll", maxRecordsPerPoll
+ ),
+ RabbitStreamSupervisorTuningConfig.class
+ );
+ }
+
}
diff --git a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaIOConfigBuilder.java b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaIOConfigBuilder.java
similarity index 68%
rename from extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaIOConfigBuilder.java
rename to extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaIOConfigBuilder.java
index ac8a9e1d2f7..ce14ef64494 100644
--- a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaIOConfigBuilder.java
+++ b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaIOConfigBuilder.java
@@ -21,7 +21,7 @@ package org.apache.druid.indexing.kafka.supervisor;
import org.apache.druid.data.input.InputFormat;
import org.apache.druid.data.input.kafkainput.KafkaInputFormat;
-import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
+import org.apache.druid.indexing.seekablestream.extension.KafkaConfigOverrides;
import org.apache.druid.indexing.seekablestream.supervisor.SupervisorIOConfigBuilder;
import java.util.Map;
@@ -34,7 +34,9 @@ public class KafkaIOConfigBuilder extends SupervisorIOConfigBuilder<KafkaIOConfi
private String topic;
private String topicPattern;
private Map<String, Object> consumerProperties;
- private BoundedStreamConfig boundedStreamConfig;
+ private Long pollTimeout;
+ private KafkaConfigOverrides configOverrides;
+ private Boolean emitTimeLagMetrics;
public KafkaIOConfigBuilder withTopic(String topic)
{
@@ -54,6 +56,24 @@ public class KafkaIOConfigBuilder extends SupervisorIOConfigBuilder<KafkaIOConfi
return this;
}
+ public KafkaIOConfigBuilder withPollTimeout(Long pollTimeout)
+ {
+ this.pollTimeout = pollTimeout;
+ return this;
+ }
+
+ public KafkaIOConfigBuilder withConfigOverrides(KafkaConfigOverrides configOverrides)
+ {
+ this.configOverrides = configOverrides;
+ return this;
+ }
+
+ public KafkaIOConfigBuilder withEmitTimeLagMetrics(Boolean emitTimeLagMetrics)
+ {
+ this.emitTimeLagMetrics = emitTimeLagMetrics;
+ return this;
+ }
+
public KafkaIOConfigBuilder withKafkaInputFormat(InputFormat valueFormat)
{
this.inputFormat = new KafkaInputFormat(
@@ -70,9 +90,18 @@ public class KafkaIOConfigBuilder extends SupervisorIOConfigBuilder<KafkaIOConfi
return this;
}
- public KafkaIOConfigBuilder withBoundedStreamConfig(BoundedStreamConfig boundedStreamConfig)
+ /**
+ * Populates this builder (base and Kafka-specific fields) from an existing config.
+ */
+ public KafkaIOConfigBuilder copyFrom(KafkaSupervisorIOConfig io)
{
- this.boundedStreamConfig = boundedStreamConfig;
+ copyFromBase(io);
+ this.topic = io.getTopic();
+ this.topicPattern = io.getTopicPattern();
+ this.consumerProperties = io.getConsumerProperties();
+ this.pollTimeout = io.getPollTimeout();
+ this.configOverrides = io.getConfigOverrides();
+ this.emitTimeLagMetrics = io.isEmitTimeLagMetrics();
return this;
}
@@ -89,7 +118,7 @@ public class KafkaIOConfigBuilder extends SupervisorIOConfigBuilder<KafkaIOConfi
consumerProperties,
autoScalerConfig,
lagAggregator,
- null,
+ pollTimeout,
startDelay,
period,
useEarliestSequenceNumber,
@@ -97,11 +126,11 @@ public class KafkaIOConfigBuilder extends SupervisorIOConfigBuilder<KafkaIOConfi
lateMessageRejectionPeriod,
earlyMessageRejectionPeriod,
lateMessageRejectionStartDateTime,
- null,
+ configOverrides,
idleConfig,
stopTaskCount,
- null,
- null,
+ emitTimeLagMetrics,
+ serverPriorityToReplicas,
boundedStreamConfig
);
}
diff --git a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java
index 7e9fffb7e79..fd844b61ba1 100644
--- a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java
+++ b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfig.java
@@ -37,6 +37,7 @@ import org.joda.time.Period;
import javax.annotation.Nullable;
import java.util.Map;
+import java.util.Objects;
public class KafkaSupervisorIOConfig extends SeekableStreamSupervisorIOConfig
{
@@ -212,4 +213,42 @@ public class KafkaSupervisorIOConfig extends SeekableStreamSupervisorIOConfig
return topic != null ? topic : topicPattern;
}
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ KafkaSupervisorIOConfig that = (KafkaSupervisorIOConfig) o;
+ return pollTimeout == that.pollTimeout
+ && emitTimeLagMetrics == that.emitTimeLagMetrics
+ && Objects.equals(consumerProperties, that.consumerProperties)
+ && Objects.equals(configOverrides, that.configOverrides)
+ && Objects.equals(topic, that.topic)
+ && Objects.equals(topicPattern, that.topicPattern);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(
+ super.hashCode(),
+ consumerProperties,
+ pollTimeout,
+ configOverrides,
+ topic,
+ topicPattern,
+ emitTimeLagMetrics
+ );
+ }
+
+ @Override
+ public KafkaIOConfigBuilder toBuilder()
+ {
+ return new KafkaIOConfigBuilder().copyFrom(this);
+ }
+
}
diff --git a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java
index 31d3e8fad69..eb29ca58d0f 100644
--- a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java
+++ b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorSpec.java
@@ -267,4 +267,42 @@ public class KafkaSupervisorSpec extends SeekableStreamSupervisorSpec
", suspend=" + isSuspended() +
'}';
}
+
+ @Override
+ public Builder toBuilder()
+ {
+ return new Builder().copyFrom(this);
+ }
+
+ public static class Builder extends SeekableStreamSupervisorSpec.Builder<Builder>
+ {
+ @Override
+ protected Builder self()
+ {
+ return this;
+ }
+
+ @Override
+ public KafkaSupervisorSpec build()
+ {
+ return new KafkaSupervisorSpec(
+ id,
+ null,
+ dataSchema,
+ (KafkaSupervisorTuningConfig) tuningConfig,
+ (KafkaSupervisorIOConfig) ioConfig,
+ context,
+ suspended,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null
+ );
+ }
+ }
}
diff --git a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTuningConfig.java b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTuningConfig.java
index f08f652e4bd..7e6cbaf0d87 100644
--- a/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTuningConfig.java
+++ b/extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTuningConfig.java
@@ -30,6 +30,7 @@ import org.joda.time.Duration;
import org.joda.time.Period;
import javax.annotation.Nullable;
+import java.util.Objects;
public class KafkaSupervisorTuningConfig extends KafkaIndexTaskTuningConfig
implements SeekableStreamSupervisorTuningConfig
@@ -216,6 +217,29 @@ public class KafkaSupervisorTuningConfig extends KafkaIndexTaskTuningConfig
'}';
}
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ KafkaSupervisorTuningConfig that = (KafkaSupervisorTuningConfig) o;
+ return Objects.equals(workerThreads, that.workerThreads)
+ && Objects.equals(chatRetries, that.chatRetries)
+ && Objects.equals(httpTimeout, that.httpTimeout)
+ && Objects.equals(shutdownTimeout, that.shutdownTimeout)
+ && Objects.equals(offsetFetchPeriod, that.offsetFetchPeriod);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(super.hashCode(), workerThreads, chatRetries, httpTimeout, shutdownTimeout, offsetFetchPeriod);
+ }
+
@Override
public KafkaIndexTaskTuningConfig convertToTaskTuningConfig()
{
diff --git a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfigTest.java b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfigTest.java
index a6a01d8640d..0bf2e7a30e2 100644
--- a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfigTest.java
+++ b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorIOConfigTest.java
@@ -22,14 +22,20 @@ package org.apache.druid.indexing.kafka.supervisor;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
import com.google.common.collect.ImmutableMap;
+import nl.jqno.equalsverifier.EqualsVerifier;
+import nl.jqno.equalsverifier.Warning;
+import org.apache.druid.data.input.InputFormat;
import org.apache.druid.error.DruidException;
import org.apache.druid.indexing.kafka.KafkaConsumerConfigs;
import org.apache.druid.indexing.kafka.KafkaIndexTaskModule;
import org.apache.druid.indexing.kafka.KafkaRecordSupplier;
+import org.apache.druid.indexing.seekablestream.extension.KafkaConfigOverrides;
import org.apache.druid.indexing.seekablestream.supervisor.BoundedStreamConfig;
import org.apache.druid.indexing.seekablestream.supervisor.IdleConfig;
import org.apache.druid.indexing.seekablestream.supervisor.LagAggregator;
+import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.AutoScalerConfig;
import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.LagBasedAutoScalerConfig;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
@@ -45,6 +51,8 @@ import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
+import static org.easymock.EasyMock.createMock;
+
public class KafkaSupervisorIOConfigTest
{
private final ObjectMapper mapper;
@@ -618,4 +626,79 @@ public class KafkaSupervisorIOConfigTest
Assert.assertEquals(2, deserialized.getBoundedStreamConfig().getStartSequenceNumbers().size());
Assert.assertEquals(2, deserialized.getBoundedStreamConfig().getEndSequenceNumbers().size());
}
+
+ private static KafkaIOConfigBuilder ioConfigBuilder()
+ {
+ return new KafkaIOConfigBuilder()
+ .withTopic("topic")
+ .withConsumerProperties(Map.of("bootstrap.servers", "localhost:9092"))
+ .withReplicas(1)
+ .withTaskCount(2)
+ .withTaskDuration(new Period("PT1H"));
+ }
+
+ @Test
+ public void testEqualsAndHashCode()
+ {
+ final KafkaSupervisorIOConfig config = ioConfigBuilder().build();
+ Assert.assertEquals(config, ioConfigBuilder().build());
+ Assert.assertEquals(config.hashCode(), ioConfigBuilder().build().hashCode());
+ Assert.assertNotEquals(config, null);
+ Assert.assertNotEquals(config, "not an io config");
+ Assert.assertNotEquals(config, ioConfigBuilder().withTopic("other").build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withReplicas(9).build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withTaskCount(9).build());
+ Assert.assertNotEquals(
+ config,
+ ioConfigBuilder().withConsumerProperties(Map.of("bootstrap.servers", "other:9092")).build()
+ );
+ Assert.assertNotEquals(config, ioConfigBuilder().withEmitTimeLagMetrics(true).build());
+ }
+
+ @Test
+ public void testTuningConfigEqualsAndHashCode()
+ {
+ final KafkaSupervisorTuningConfig config = new KafkaTuningConfigBuilder().build();
+ Assert.assertEquals(config, new KafkaTuningConfigBuilder().build());
+ Assert.assertEquals(config.hashCode(), new KafkaTuningConfigBuilder().build().hashCode());
+ Assert.assertNotEquals(config, null);
+ Assert.assertNotEquals(config, "not a tuning config");
+ Assert.assertNotEquals(config, new KafkaTuningConfigBuilder().withWorkerThreads(99).build());
+ Assert.assertNotEquals(config, new KafkaTuningConfigBuilder().withShutdownTimeout(new Period("PT99M")).build());
+ Assert.assertNotEquals(config, new KafkaTuningConfigBuilder().withOffsetFetchPeriod(new Period("PT99S")).build());
+ }
+
+ /**
+ * Drift guard for this class's own fields (base fields are covered by SeekableStreamSupervisorIOConfigTest):
+ * a field omitted from {@code equals} would let a changed spec persist without restarting the supervisor.
+ */
+ @Test
+ public void testEqualsContractCoversAllFields()
+ {
+ EqualsVerifier.forClass(KafkaSupervisorIOConfig.class)
+ .usingGetClass()
+ .withRedefinedSuperclass()
+ .withIgnoredFields("taskCountExplicit", "autoScalerEnabled")
+ .suppress(Warning.NONFINAL_FIELDS)
+ .withPrefabValues(Optional.class, Optional.of("a"), Optional.of("b"))
+ .withPrefabValues(InputFormat.class, createMock(InputFormat.class), createMock(InputFormat.class))
+ .withPrefabValues(AutoScalerConfig.class, createMock(AutoScalerConfig.class), createMock(AutoScalerConfig.class))
+ .withPrefabValues(LagAggregator.class, createMock(LagAggregator.class), createMock(LagAggregator.class))
+ .withPrefabValues(
+ KafkaConfigOverrides.class,
+ createMock(KafkaConfigOverrides.class),
+ createMock(KafkaConfigOverrides.class)
+ )
+ .verify();
+ }
+
+ @Test
+ public void testTuningConfigEqualsContractCoversAllFields()
+ {
+ EqualsVerifier.forClass(KafkaSupervisorTuningConfig.class)
+ .usingGetClass()
+ .withRedefinedSuperclass()
+ .suppress(Warning.NONFINAL_FIELDS)
+ .verify();
+ }
}
diff --git a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java
index 65c60c38418..a954b644e42 100644
--- a/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java
+++ b/extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java
@@ -5445,34 +5445,19 @@ public class KafkaSupervisorTest extends EasyMockSupport
final Map<String, Object> consumerProperties = KafkaConsumerConfigs.getConsumerProperties();
consumerProperties.put("bootstrap.servers", kafkaHost);
- final KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaSupervisorIOConfig(
- topic,
- null,
- INPUT_FORMAT,
- null,
- 1,
- new Period("PT1H"),
- consumerProperties,
- null,
- null,
- KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
- new Period("P1D"),
- new Period("PT30S"),
- true,
- new Period("PT30M"),
- null,
- null,
- null,
- null,
- null,
- null,
- true,
- Map.of(
- 10, 2,
- 20, 3
- ),
- null
- );
+ final KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaIOConfigBuilder()
+ .withTopic(topic)
+ .withInputFormat(INPUT_FORMAT)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withConsumerProperties(consumerProperties)
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(true)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withEmitTimeLagMetrics(true)
+ .withServerPriorityToReplicas(Map.of(10, 2, 20, 3))
+ .build();
Assert.assertEquals(5, (int) kafkaSupervisorIOConfig.getReplicas());
@@ -5563,31 +5548,20 @@ public class KafkaSupervisorTest extends EasyMockSupport
final Map<String, Object> consumerProperties = KafkaConsumerConfigs.getConsumerProperties();
consumerProperties.put("bootstrap.servers", kafkaHost);
- final KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaSupervisorIOConfig(
- topic,
- null,
- INPUT_FORMAT,
- 1,
- 1,
- new Period("PT1H"),
- consumerProperties,
- null,
- null,
- KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- null,
- null,
- null,
- true,
- null,
- new BoundedStreamConfig(startOffsets, endOffsets)
- );
+ final KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaIOConfigBuilder()
+ .withTopic(topic)
+ .withInputFormat(INPUT_FORMAT)
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withConsumerProperties(consumerProperties)
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withEmitTimeLagMetrics(true)
+ .withBoundedStreamConfig(new BoundedStreamConfig(startOffsets, endOffsets))
+ .build();
Assert.assertTrue(kafkaSupervisorIOConfig.isBounded());
@@ -5840,31 +5814,24 @@ public class KafkaSupervisorTest extends EasyMockSupport
final Map<String, Object> consumerProperties = KafkaConsumerConfigs.getConsumerProperties();
consumerProperties.put("myCustomKey", "myCustomValue");
consumerProperties.put("bootstrap.servers", kafkaHost);
- KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaSupervisorIOConfig(
- multiTopic ? null : topic,
- multiTopic ? topicPattern : null,
- INPUT_FORMAT,
- replicas,
- taskCount,
- new Period(duration),
- consumerProperties,
- null,
- null,
- KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
- new Period("P1D"),
- new Period("PT30S"),
- useEarliestOffset,
- new Period("PT30M"),
- lateMessageRejectionPeriod,
- earlyMessageRejectionPeriod,
- null,
- null,
- idleConfig,
- null,
- true,
- serverPriorityToReplicas,
- null
- );
+ KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaIOConfigBuilder()
+ .withTopic(multiTopic ? null : topic)
+ .withTopicPattern(multiTopic ? topicPattern : null)
+ .withInputFormat(INPUT_FORMAT)
+ .withReplicas(replicas)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period(duration))
+ .withConsumerProperties(consumerProperties)
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(useEarliestOffset)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLateMessageRejectionPeriod(lateMessageRejectionPeriod)
+ .withEarlyMessageRejectionPeriod(earlyMessageRejectionPeriod)
+ .withIdleConfig(idleConfig)
+ .withEmitTimeLagMetrics(true)
+ .withServerPriorityToReplicas(serverPriorityToReplicas)
+ .build();
KafkaIndexTaskClientFactory taskClientFactory = new KafkaIndexTaskClientFactory(
null,
@@ -5936,31 +5903,20 @@ public class KafkaSupervisorTest extends EasyMockSupport
consumerProperties.put("myCustomKey", "myCustomValue");
consumerProperties.put("bootstrap.servers", kafkaHost);
consumerProperties.put("isolation.level", "read_committed");
- KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaSupervisorIOConfig(
- topic,
- null,
- INPUT_FORMAT,
- replicas,
- taskCount,
- new Period(duration),
- consumerProperties,
- null,
- null,
- KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
- new Period("P1D"),
- new Period("PT30S"),
- useEarliestOffset,
- new Period("PT30M"),
- lateMessageRejectionPeriod,
- earlyMessageRejectionPeriod,
- null,
- null,
- null,
- null,
- false,
- null,
- null
- );
+ KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaIOConfigBuilder()
+ .withTopic(topic)
+ .withInputFormat(INPUT_FORMAT)
+ .withReplicas(replicas)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period(duration))
+ .withConsumerProperties(consumerProperties)
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(useEarliestOffset)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLateMessageRejectionPeriod(lateMessageRejectionPeriod)
+ .withEarlyMessageRejectionPeriod(earlyMessageRejectionPeriod)
+ .build();
KafkaIndexTaskClientFactory taskClientFactory = new KafkaIndexTaskClientFactory(
null,
@@ -6032,31 +5988,20 @@ public class KafkaSupervisorTest extends EasyMockSupport
consumerProperties.put("myCustomKey", "myCustomValue");
consumerProperties.put("bootstrap.servers", kafkaHost);
consumerProperties.put("isolation.level", "read_committed");
- KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaSupervisorIOConfig(
- topic,
- null,
- INPUT_FORMAT,
- replicas,
- taskCount,
- new Period(duration),
- consumerProperties,
- null,
- null,
- KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
- new Period("P1D"),
- new Period("PT30S"),
- useEarliestOffset,
- new Period("PT30M"),
- lateMessageRejectionPeriod,
- earlyMessageRejectionPeriod,
- null,
- null,
- null,
- null,
- false,
- null,
- null
- );
+ KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaIOConfigBuilder()
+ .withTopic(topic)
+ .withInputFormat(INPUT_FORMAT)
+ .withReplicas(replicas)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period(duration))
+ .withConsumerProperties(consumerProperties)
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(useEarliestOffset)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLateMessageRejectionPeriod(lateMessageRejectionPeriod)
+ .withEarlyMessageRejectionPeriod(earlyMessageRejectionPeriod)
+ .build();
KafkaIndexTaskClientFactory taskClientFactory = new KafkaIndexTaskClientFactory(
null,
@@ -6610,31 +6555,20 @@ public class KafkaSupervisorTest extends EasyMockSupport
final Map<String, Object> consumerProperties = KafkaConsumerConfigs.getConsumerProperties();
consumerProperties.put("myCustomKey", "myCustomValue");
consumerProperties.put("bootstrap.servers", kafkaHost);
- KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaSupervisorIOConfig(
- topic,
- null,
- INPUT_FORMAT,
- replicas,
- taskCount,
- new Period(duration),
- consumerProperties,
- null,
- null,
- KafkaSupervisorIOConfig.DEFAULT_POLL_TIMEOUT_MILLIS,
- new Period("P1D"),
- new Period("PT30S"),
- true,
- new Period("PT30M"),
- null,
- null,
- null,
- null,
- null,
- null,
- true,
- null,
- boundedConfig
- );
+ KafkaSupervisorIOConfig kafkaSupervisorIOConfig = new KafkaIOConfigBuilder()
+ .withTopic(topic)
+ .withInputFormat(INPUT_FORMAT)
+ .withReplicas(replicas)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period(duration))
+ .withConsumerProperties(consumerProperties)
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(true)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withEmitTimeLagMetrics(true)
+ .withBoundedStreamConfig(boundedConfig)
+ .build();
KafkaIndexTaskClientFactory taskClientFactory = new KafkaIndexTaskClientFactory(
null,
diff --git a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisIOConfigBuilder.java b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisIOConfigBuilder.java
new file mode 100644
index 00000000000..8296f012dc0
--- /dev/null
+++ b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisIOConfigBuilder.java
@@ -0,0 +1,125 @@
+/*
+ * 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.kinesis.supervisor;
+
+import org.apache.druid.indexing.kinesis.KinesisRegion;
+import org.apache.druid.indexing.seekablestream.supervisor.SupervisorIOConfigBuilder;
+
+/**
+ * Builder for {@link KinesisSupervisorIOConfig}.
+ */
+public class KinesisIOConfigBuilder extends SupervisorIOConfigBuilder<KinesisIOConfigBuilder, KinesisSupervisorIOConfig>
+{
+ private String endpoint;
+ private KinesisRegion region;
+ private Integer recordsPerFetch;
+ private Integer fetchDelayMillis;
+ private String awsAssumedRoleArn;
+ private String awsExternalId;
+ private Boolean deaggregate;
+
+ public KinesisIOConfigBuilder withEndpoint(String endpoint)
+ {
+ this.endpoint = endpoint;
+ return this;
+ }
+
+ public KinesisIOConfigBuilder withRegion(KinesisRegion region)
+ {
+ this.region = region;
+ return this;
+ }
+
+ public KinesisIOConfigBuilder withRecordsPerFetch(Integer recordsPerFetch)
+ {
+ this.recordsPerFetch = recordsPerFetch;
+ return this;
+ }
+
+ public KinesisIOConfigBuilder withFetchDelayMillis(Integer fetchDelayMillis)
+ {
+ this.fetchDelayMillis = fetchDelayMillis;
+ return this;
+ }
+
+ public KinesisIOConfigBuilder withAwsAssumedRoleArn(String awsAssumedRoleArn)
+ {
+ this.awsAssumedRoleArn = awsAssumedRoleArn;
+ return this;
+ }
+
+ public KinesisIOConfigBuilder withAwsExternalId(String awsExternalId)
+ {
+ this.awsExternalId = awsExternalId;
+ return this;
+ }
+
+ public KinesisIOConfigBuilder withDeaggregate(Boolean deaggregate)
+ {
+ this.deaggregate = deaggregate;
+ return this;
+ }
+
+ /**
+ * Populates this builder (base and Kinesis-specific fields) from an existing config. The
+ * resolved {@code endpoint} is copied directly, so {@code region} (only used to derive an
+ * endpoint) is left null.
+ */
+ public KinesisIOConfigBuilder copyFrom(KinesisSupervisorIOConfig io)
+ {
+ copyFromBase(io);
+ this.endpoint = io.getEndpoint();
+ this.recordsPerFetch = io.getRecordsPerFetch();
+ this.fetchDelayMillis = io.getFetchDelayMillis();
+ this.awsAssumedRoleArn = io.getAwsAssumedRoleArn();
+ this.awsExternalId = io.getAwsExternalId();
+ this.deaggregate = io.isDeaggregate();
+ return this;
+ }
+
+ @Override
+ public KinesisSupervisorIOConfig build()
+ {
+ return new KinesisSupervisorIOConfig(
+ stream,
+ inputFormat,
+ endpoint,
+ region,
+ replicas,
+ taskCount,
+ taskDuration,
+ startDelay,
+ period,
+ useEarliestSequenceNumber,
+ completionTimeout,
+ lateMessageRejectionPeriod,
+ earlyMessageRejectionPeriod,
+ lateMessageRejectionStartDateTime,
+ recordsPerFetch,
+ fetchDelayMillis,
+ awsAssumedRoleArn,
+ awsExternalId,
+ autoScalerConfig,
+ deaggregate != null && deaggregate,
+ serverPriorityToReplicas,
+ boundedStreamConfig
+ );
+ }
+}
diff --git a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorIOConfig.java b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorIOConfig.java
index e20d3a261cf..93b64efd902 100644
--- a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorIOConfig.java
+++ b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorIOConfig.java
@@ -36,6 +36,7 @@ import org.joda.time.Period;
import javax.annotation.Nullable;
import java.util.Map;
+import java.util.Objects;
public class KinesisSupervisorIOConfig extends SeekableStreamSupervisorIOConfig
{
@@ -183,4 +184,42 @@ public class KinesisSupervisorIOConfig extends SeekableStreamSupervisorIOConfig
", deaggregate=" + deaggregate +
'}';
}
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ KinesisSupervisorIOConfig that = (KinesisSupervisorIOConfig) o;
+ return fetchDelayMillis == that.fetchDelayMillis
+ && deaggregate == that.deaggregate
+ && Objects.equals(endpoint, that.endpoint)
+ && Objects.equals(recordsPerFetch, that.recordsPerFetch)
+ && Objects.equals(awsAssumedRoleArn, that.awsAssumedRoleArn)
+ && Objects.equals(awsExternalId, that.awsExternalId);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(
+ super.hashCode(),
+ endpoint,
+ recordsPerFetch,
+ fetchDelayMillis,
+ awsAssumedRoleArn,
+ awsExternalId,
+ deaggregate
+ );
+ }
+
+ @Override
+ public KinesisIOConfigBuilder toBuilder()
+ {
+ return new KinesisIOConfigBuilder().copyFrom(this);
+ }
}
diff --git a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
index 4899337797b..215aade5c8c 100644
--- a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
+++ b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorSpec.java
@@ -171,6 +171,53 @@ public class KinesisSupervisorSpec extends SeekableStreamSupervisorSpec
return (KinesisSupervisorIngestionSpec) super.getSpec();
}
+ @Override
+ public Builder toBuilder()
+ {
+ return new Builder().awsCredentialsConfig(awsCredentialsConfig).copyFrom(this);
+ }
+
+ public static class Builder extends SeekableStreamSupervisorSpec.Builder<Builder>
+ {
+ private AWSCredentialsConfig awsCredentialsConfig;
+
+ public Builder awsCredentialsConfig(AWSCredentialsConfig awsCredentialsConfig)
+ {
+ this.awsCredentialsConfig = awsCredentialsConfig;
+ return this;
+ }
+
+ @Override
+ protected Builder self()
+ {
+ return this;
+ }
+
+ @Override
+ public KinesisSupervisorSpec build()
+ {
+ return new KinesisSupervisorSpec(
+ id,
+ null,
+ dataSchema,
+ (KinesisSupervisorTuningConfig) tuningConfig,
+ (KinesisSupervisorIOConfig) ioConfig,
+ context,
+ suspended,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ awsCredentialsConfig,
+ null
+ );
+ }
+ }
+
@Override
protected KinesisSupervisorSpec toggleSuspend(boolean suspend)
{
diff --git a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTuningConfig.java b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTuningConfig.java
index 1a11f8d658b..b3643e9bff4 100644
--- a/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTuningConfig.java
+++ b/extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTuningConfig.java
@@ -29,6 +29,7 @@ import org.joda.time.Duration;
import org.joda.time.Period;
import javax.annotation.Nullable;
+import java.util.Objects;
public class KinesisSupervisorTuningConfig extends KinesisIndexTaskTuningConfig
implements SeekableStreamSupervisorTuningConfig
@@ -214,6 +215,40 @@ public class KinesisSupervisorTuningConfig extends KinesisIndexTaskTuningConfig
return useListShards;
}
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (!super.equals(o)) {
+ return false;
+ }
+ KinesisSupervisorTuningConfig that = (KinesisSupervisorTuningConfig) o;
+ return useListShards == that.useListShards
+ && Objects.equals(workerThreads, that.workerThreads)
+ && Objects.equals(chatRetries, that.chatRetries)
+ && Objects.equals(httpTimeout, that.httpTimeout)
+ && Objects.equals(shutdownTimeout, that.shutdownTimeout)
+ && Objects.equals(repartitionTransitionDuration, that.repartitionTransitionDuration)
+ && Objects.equals(offsetFetchPeriod, that.offsetFetchPeriod);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(
+ super.hashCode(),
+ workerThreads,
+ chatRetries,
+ httpTimeout,
+ shutdownTimeout,
+ repartitionTransitionDuration,
+ offsetFetchPeriod,
+ useListShards
+ );
+ }
+
@Override
public String toString()
{
diff --git a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisSamplerSpecTest.java b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisSamplerSpecTest.java
index 6f4410f3a8b..5513a7ddd2f 100644
--- a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisSamplerSpecTest.java
+++ b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisSamplerSpecTest.java
@@ -33,7 +33,7 @@ import org.apache.druid.data.input.impl.StringDimensionSchema;
import org.apache.druid.data.input.impl.TimestampSpec;
import org.apache.druid.data.input.kinesis.KinesisRecordEntity;
import org.apache.druid.indexer.granularity.UniformGranularitySpec;
-import org.apache.druid.indexing.kinesis.supervisor.KinesisSupervisorIOConfig;
+import org.apache.druid.indexing.kinesis.supervisor.KinesisIOConfigBuilder;
import org.apache.druid.indexing.kinesis.supervisor.KinesisSupervisorSpec;
import org.apache.druid.indexing.overlord.sampler.InputSourceSampler;
import org.apache.druid.indexing.overlord.sampler.SamplerConfig;
@@ -62,6 +62,7 @@ import java.time.Instant;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
+import java.util.Map;
public class KinesisSamplerSpecTest extends EasyMockSupport
{
@@ -119,30 +120,11 @@ public class KinesisSamplerSpecTest extends EasyMockSupport
null,
DATA_SCHEMA,
null,
- new KinesisSupervisorIOConfig(
- STREAM,
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- true,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- false,
- null,
- null
- ),
+ new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withUseEarliestSequenceNumber(true)
+ .build(),
null,
null,
null,
@@ -175,30 +157,11 @@ public class KinesisSamplerSpecTest extends EasyMockSupport
null,
DATA_SCHEMA,
null,
- new KinesisSupervisorIOConfig(
- STREAM,
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- true,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- false,
- null,
- null
- ),
+ new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withUseEarliestSequenceNumber(true)
+ .build(),
null,
null,
null,
diff --git a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorIOConfigTest.java b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorIOConfigTest.java
index a93bdf6871b..9ea1f71ed15 100644
--- a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorIOConfigTest.java
+++ b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorIOConfigTest.java
@@ -21,16 +21,25 @@ package org.apache.druid.indexing.kinesis.supervisor;
import com.fasterxml.jackson.databind.JsonMappingException;
import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Optional;
+import nl.jqno.equalsverifier.EqualsVerifier;
+import nl.jqno.equalsverifier.Warning;
+import org.apache.druid.data.input.InputFormat;
import org.apache.druid.indexing.kinesis.KinesisIndexingServiceModule;
import org.apache.druid.indexing.kinesis.KinesisRegion;
+import org.apache.druid.indexing.seekablestream.supervisor.LagAggregator;
+import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.AutoScalerConfig;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.hamcrest.CoreMatchers;
import org.joda.time.Duration;
+import org.joda.time.Period;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
+import static org.easymock.EasyMock.createMock;
+
public class KinesisSupervisorIOConfigTest
{
private final ObjectMapper mapper;
@@ -204,4 +213,68 @@ public class KinesisSupervisorIOConfigTest
Assert.assertNull(config.getBoundedStreamConfig());
}
+ private static KinesisIOConfigBuilder ioConfigBuilder()
+ {
+ return new KinesisIOConfigBuilder()
+ .withStream("stream")
+ .withEndpoint("awsEndpoint")
+ .withReplicas(1)
+ .withTaskCount(2)
+ .withTaskDuration(new Period("PT1H"));
+ }
+
+ @Test
+ public void testEqualsAndHashCode()
+ {
+ final KinesisSupervisorIOConfig config = ioConfigBuilder().build();
+ Assert.assertEquals(config, ioConfigBuilder().build());
+ Assert.assertEquals(config.hashCode(), ioConfigBuilder().build().hashCode());
+ Assert.assertNotEquals(config, null);
+ Assert.assertNotEquals(config, "not an io config");
+ Assert.assertNotEquals(config, ioConfigBuilder().withEndpoint("other").build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withReplicas(9).build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withTaskCount(9).build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withFetchDelayMillis(999).build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withDeaggregate(true).build());
+ }
+
+ @Test
+ public void testTuningConfigEqualsAndHashCode()
+ {
+ final KinesisSupervisorTuningConfig config = KinesisSupervisorTuningConfig.defaultConfig();
+ Assert.assertEquals(config, KinesisSupervisorTuningConfig.defaultConfig());
+ Assert.assertEquals(config.hashCode(), KinesisSupervisorTuningConfig.defaultConfig().hashCode());
+ Assert.assertNotEquals(config, null);
+ Assert.assertNotEquals(config, "not a tuning config");
+ }
+
+ /**
+ * Drift guard for this class's own fields (base fields are covered by SeekableStreamSupervisorIOConfigTest):
+ * a field omitted from {@code equals} would let a changed spec persist without restarting the supervisor.
+ */
+ @Test
+ public void testEqualsContractCoversAllFields()
+ {
+ EqualsVerifier.forClass(KinesisSupervisorIOConfig.class)
+ .usingGetClass()
+ .withRedefinedSuperclass()
+ .withIgnoredFields("taskCountExplicit", "autoScalerEnabled")
+ .suppress(Warning.NONFINAL_FIELDS)
+ .withPrefabValues(Optional.class, Optional.of("a"), Optional.of("b"))
+ .withPrefabValues(InputFormat.class, createMock(InputFormat.class), createMock(InputFormat.class))
+ .withPrefabValues(AutoScalerConfig.class, createMock(AutoScalerConfig.class), createMock(AutoScalerConfig.class))
+ .withPrefabValues(LagAggregator.class, createMock(LagAggregator.class), createMock(LagAggregator.class))
+ .verify();
+ }
+
+ @Test
+ public void testTuningConfigEqualsContractCoversAllFields()
+ {
+ EqualsVerifier.forClass(KinesisSupervisorTuningConfig.class)
+ .usingGetClass()
+ .withRedefinedSuperclass()
+ .suppress(Warning.NONFINAL_FIELDS)
+ .verify();
+ }
+
}
diff --git a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTest.java b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTest.java
index 3f6da8aa053..2c6f8a24502 100644
--- a/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTest.java
+++ b/extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTest.java
@@ -448,30 +448,20 @@ public class KinesisSupervisorTest extends EasyMockSupport
@Test
public void testRecordSupplier()
{
- KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisSupervisorIOConfig(
- STREAM,
- INPUT_FORMAT,
- "awsEndpoint",
- null,
- 1,
- 1,
- new Period("PT30M"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- 100,
- 1000,
- null,
- null,
- null,
- false,
- null,
- null
- );
+ KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(INPUT_FORMAT)
+ .withEndpoint("awsEndpoint")
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT30M"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withRecordsPerFetch(100)
+ .withFetchDelayMillis(1000)
+ .build();
KinesisIndexTaskClientFactory clientFactory = new KinesisIndexTaskClientFactory(null, OBJECT_MAPPER);
KinesisSupervisor supervisor = new KinesisSupervisor(
taskStorage,
@@ -515,59 +505,40 @@ public class KinesisSupervisorTest extends EasyMockSupport
public void testKinesisIOConfigInitAndAutoscalerConfigCreation()
{
// create KinesisSupervisorIOConfig with autoScalerConfig null
- KinesisSupervisorIOConfig kinesisSupervisorIOConfigWithNullAutoScalerConfig = new KinesisSupervisorIOConfig(
- STREAM,
- INPUT_FORMAT,
- "awsEndpoint",
- null,
- 1,
- 1,
- new Period("PT30M"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- 100,
- 1000,
- null,
- null,
- null,
- false,
- null,
- null
- );
+ KinesisSupervisorIOConfig kinesisSupervisorIOConfigWithNullAutoScalerConfig = new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(INPUT_FORMAT)
+ .withEndpoint("awsEndpoint")
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT30M"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withRecordsPerFetch(100)
+ .withFetchDelayMillis(1000)
+ .build();
AutoScalerConfig autoscalerConfigNull = kinesisSupervisorIOConfigWithNullAutoScalerConfig.getAutoScalerConfig();
Assert.assertNull(autoscalerConfigNull);
// create KinesisSupervisorIOConfig with autoScalerConfig Empty
- KinesisSupervisorIOConfig kinesisSupervisorIOConfigWithEmptyAutoScalerConfig = new KinesisSupervisorIOConfig(
- STREAM,
- INPUT_FORMAT,
- "awsEndpoint",
- null,
- 1,
- 1,
- new Period("PT30M"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- 100,
- 1000,
- null,
- null,
- OBJECT_MAPPER.convertValue(new HashMap<>(), AutoScalerConfig.class),
- false,
- null,
- null
- );
+ KinesisSupervisorIOConfig kinesisSupervisorIOConfigWithEmptyAutoScalerConfig = new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(INPUT_FORMAT)
+ .withEndpoint("awsEndpoint")
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT30M"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withRecordsPerFetch(100)
+ .withFetchDelayMillis(1000)
+ .withAutoScalerConfig(OBJECT_MAPPER.convertValue(new HashMap<>(), AutoScalerConfig.class))
+ .build();
AutoScalerConfig autoscalerConfig = kinesisSupervisorIOConfigWithEmptyAutoScalerConfig.getAutoScalerConfig();
Assert.assertNotNull(autoscalerConfig);
@@ -4199,30 +4170,10 @@ public class KinesisSupervisorTest extends EasyMockSupport
null,
dataSchema,
null,
- new KinesisSupervisorIOConfig(
- STREAM,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- true,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- null,
- false,
- null,
- null
- ),
+ new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withUseEarliestSequenceNumber(true)
+ .build(),
null,
null,
null,
@@ -4741,30 +4692,18 @@ public class KinesisSupervisorTest extends EasyMockSupport
"shardId-000000000000", "49590338271534258098958632799622211348320767794297876514",
"shardId-000000000001", "49590338271556258844158102930252531974520363696878520322"
);
- final KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisSupervisorIOConfig(
- STREAM,
- INPUT_FORMAT,
- "awsEndpoint",
- null,
- 1,
- 1,
- new Period("PT30S"),
- null,
- new Period("PT30M"),
- null,
- null,
- null,
- null,
- null,
- null,
- 0,
- null,
- null,
- null,
- true,
- null,
- new BoundedStreamConfig(startOffsets, endOffsets)
- );
+ final KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(INPUT_FORMAT)
+ .withEndpoint("awsEndpoint")
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT30S"))
+ .withSupervisorRunPeriod(new Period("PT30M"))
+ .withFetchDelayMillis(0)
+ .withDeaggregate(true)
+ .withBoundedStreamConfig(new BoundedStreamConfig(startOffsets, endOffsets))
+ .build();
Assert.assertTrue(kinesisSupervisorIOConfig.isBounded());
@@ -4841,30 +4780,18 @@ public class KinesisSupervisorTest extends EasyMockSupport
Map<String, Object> startOffsets = ImmutableMap.of(SHARD_ID0, singleOffset);
Map<String, Object> endOffsets = ImmutableMap.of(SHARD_ID0, singleOffset);
- final KinesisSupervisorIOConfig ioConfig = new KinesisSupervisorIOConfig(
- STREAM,
- INPUT_FORMAT,
- "awsEndpoint",
- null,
- 1,
- 1,
- new Period("PT1H"),
- null,
- new Period("PT30M"),
- null,
- null,
- null,
- null,
- null,
- null,
- 0,
- null,
- null,
- null,
- true,
- null,
- new BoundedStreamConfig(startOffsets, endOffsets)
- );
+ final KinesisSupervisorIOConfig ioConfig = new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(INPUT_FORMAT)
+ .withEndpoint("awsEndpoint")
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withSupervisorRunPeriod(new Period("PT30M"))
+ .withFetchDelayMillis(0)
+ .withDeaggregate(true)
+ .withBoundedStreamConfig(new BoundedStreamConfig(startOffsets, endOffsets))
+ .build();
final KinesisIndexTaskClientFactory taskClientFactory = new KinesisIndexTaskClientFactory(null, null);
final KinesisSupervisorSpec spec = new KinesisSupervisorSpec(
@@ -5357,30 +5284,20 @@ public class KinesisSupervisorTest extends EasyMockSupport
boolean suspended
)
{
- KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisSupervisorIOConfig(
- STREAM,
- INPUT_FORMAT,
- "awsEndpoint",
- null,
- replicas,
- taskCount,
- new Period(duration),
- new Period("P1D"),
- new Period("PT30S"),
- useEarliestOffset,
- new Period("PT30M"),
- lateMessageRejectionPeriod,
- earlyMessageRejectionPeriod,
- null,
- null,
- null,
- null,
- null,
- null,
- false,
- null,
- null
- );
+ KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(INPUT_FORMAT)
+ .withEndpoint("awsEndpoint")
+ .withReplicas(replicas)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period(duration))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(useEarliestOffset)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLateMessageRejectionPeriod(lateMessageRejectionPeriod)
+ .withEarlyMessageRejectionPeriod(earlyMessageRejectionPeriod)
+ .build();
KinesisIndexTaskClientFactory taskClientFactory = new KinesisIndexTaskClientFactory(
null,
@@ -5501,30 +5418,22 @@ public class KinesisSupervisorTest extends EasyMockSupport
AutoScalerConfig autoScalerConfig
)
{
- KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisSupervisorIOConfig(
- STREAM,
- INPUT_FORMAT,
- "awsEndpoint",
- null,
- replicas,
- taskCount,
- new Period(duration),
- new Period("P1D"),
- new Period("PT30S"),
- useEarliestOffset,
- new Period("PT30M"),
- lateMessageRejectionPeriod,
- earlyMessageRejectionPeriod,
- null,
- null,
- fetchDelayMillis,
- null,
- null,
- autoScalerConfig,
- false,
- null,
- null
- );
+ KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(INPUT_FORMAT)
+ .withEndpoint("awsEndpoint")
+ .withReplicas(replicas)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period(duration))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(useEarliestOffset)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLateMessageRejectionPeriod(lateMessageRejectionPeriod)
+ .withEarlyMessageRejectionPeriod(earlyMessageRejectionPeriod)
+ .withFetchDelayMillis(fetchDelayMillis)
+ .withAutoScalerConfig(autoScalerConfig)
+ .build();
KinesisIndexTaskClientFactory taskClientFactory = new KinesisIndexTaskClientFactory(
null,
@@ -5589,30 +5498,21 @@ public class KinesisSupervisorTest extends EasyMockSupport
boolean isTaskCurrentReturn
)
{
- KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisSupervisorIOConfig(
- STREAM,
- INPUT_FORMAT,
- "awsEndpoint",
- null,
- replicas,
- taskCount,
- new Period(duration),
- new Period("P1D"),
- new Period("PT30S"),
- useEarliestOffset,
- new Period("PT30M"),
- lateMessageRejectionPeriod,
- earlyMessageRejectionPeriod,
- null,
- null,
- fetchDelayMillis,
- null,
- null,
- null,
- false,
- null,
- null
- );
+ KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(INPUT_FORMAT)
+ .withEndpoint("awsEndpoint")
+ .withReplicas(replicas)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period(duration))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(useEarliestOffset)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLateMessageRejectionPeriod(lateMessageRejectionPeriod)
+ .withEarlyMessageRejectionPeriod(earlyMessageRejectionPeriod)
+ .withFetchDelayMillis(fetchDelayMillis)
+ .build();
KinesisIndexTaskClientFactory taskClientFactory = new KinesisIndexTaskClientFactory(
null,
@@ -5679,30 +5579,21 @@ public class KinesisSupervisorTest extends EasyMockSupport
KinesisSupervisorTuningConfig tuningConfig
)
{
- KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisSupervisorIOConfig(
- STREAM,
- INPUT_FORMAT,
- "awsEndpoint",
- null,
- replicas,
- taskCount,
- new Period(duration),
- new Period("P1D"),
- new Period("PT30S"),
- useEarliestOffset,
- new Period("PT30M"),
- lateMessageRejectionPeriod,
- earlyMessageRejectionPeriod,
- null,
- null,
- fetchDelayMillis,
- null,
- null,
- null,
- false,
- null,
- null
- );
+ KinesisSupervisorIOConfig kinesisSupervisorIOConfig = new KinesisIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(INPUT_FORMAT)
+ .withEndpoint("awsEndpoint")
+ .withReplicas(replicas)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period(duration))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(useEarliestOffset)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLateMessageRejectionPeriod(lateMessageRejectionPeriod)
+ .withEarlyMessageRejectionPeriod(earlyMessageRejectionPeriod)
+ .withFetchDelayMillis(fetchDelayMillis)
+ .build();
KinesisIndexTaskClientFactory taskClientFactory = new KinesisIndexTaskClientFactory(
null,
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/CompactionSupervisorManager.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/CompactionSupervisorManager.java
index 4482b5e688a..c588bb48265 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/CompactionSupervisorManager.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/CompactionSupervisorManager.java
@@ -73,8 +73,8 @@ public class CompactionSupervisorManager
)
{
return performIfLeader(manager -> {
- // Check if the spec needs to be updated
- if (manager.shouldUpdateSupervisor(spec) && manager.createOrUpdateAndStartSupervisor(spec)) {
+ final SupervisorSpecUpdateResult result = manager.createOrUpdateAndStartSupervisor(spec, true);
+ if (result.isModified()) {
final String auditPayload
= StringUtils.format("Update supervisor[%s] for datasource[%s]", spec.getId(), spec.getDataSources());
auditManager.doAudit(
@@ -87,7 +87,6 @@ public class CompactionSupervisorManager
.build()
);
}
-
return true;
});
}
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 d20fb3cfe79..9f599f3aee8 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
@@ -177,12 +177,14 @@ public class SupervisorManager implements SupervisorStatsProvider
}
/**
- * Creates or updates a supervisor and then starts it.
- * If no change has been made to the supervisor spec, it is only restarted.
- *
- * @return true if the supervisor was updated, false otherwise
+ * 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).
*/
- public boolean createOrUpdateAndStartSupervisor(SupervisorSpec spec)
+ public SupervisorSpecUpdateResult createOrUpdateAndStartSupervisor(
+ final SupervisorSpec spec,
+ final boolean skipRestartIfUnmodified
+ )
{
Preconditions.checkState(started, "SupervisorManager not started");
Preconditions.checkNotNull(spec, "spec");
@@ -191,47 +193,81 @@ public class SupervisorManager implements SupervisorStatsProvider
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 = isSpecChangedAndValidated(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) {
+ // 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);
+ }
+
+ final SupervisorSpec currentSpec = current.rhs;
+ if (!areSpecBytesChanged(spec, currentSpec)) {
+ return SupervisorSpecUpdateResult.of(false, false);
+ }
+
+ // 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);
+ }
+
+ // The effective (merged) spec is what will be persisted, so validate that transition exactly once.
+ currentSpec.validateSpecUpdateTo(spec);
+
+ if (!currentSpec.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);
}
}
- /**
- * 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
- */
- public boolean shouldUpdateSupervisor(SupervisorSpec spec)
+ /** Byte-level diff against the running spec; validates allowed updates when bytes differ. */
+ private boolean isSpecChangedAndValidated(SupervisorSpec spec)
{
- Preconditions.checkState(started, "SupervisorManager not started");
- Preconditions.checkNotNull(spec, "spec");
- Preconditions.checkNotNull(spec.getId(), "spec.getId()");
- Preconditions.checkNotNull(spec.getDataSources(), "spec.getDatasources()");
- synchronized (lock) {
- Preconditions.checkState(started, "SupervisorManager not started");
- try {
- byte[] specAsBytes = jsonMapper.writeValueAsBytes(spec);
- Pair<Supervisor, SupervisorSpec> currentSupervisor = supervisors.get(spec.getId());
- if (currentSupervisor == null || currentSupervisor.rhs == null) {
- return true;
- } else if (Arrays.equals(specAsBytes, jsonMapper.writeValueAsBytes(currentSupervisor.rhs))) {
- return false;
- } else {
- // The spec bytes are different, so we need to check if the update is allowed
- currentSupervisor.rhs.validateSpecUpdateTo(spec);
- return true;
- }
- }
- catch (JsonProcessingException ex) {
- log.warn("Failed to write spec as bytes for spec_id[%s]", spec.getId());
- }
+ final Pair<Supervisor, SupervisorSpec> current = supervisors.get(spec.getId());
+ final SupervisorSpec currentSpec = current == null ? null : current.rhs;
+ if (currentSpec == null) {
+ return true;
+ } else if (!areSpecBytesChanged(spec, currentSpec)) {
+ return false;
+ } else {
+ currentSpec.validateSpecUpdateTo(spec);
+ return true;
+ }
+ }
+
+ /** Byte-level diff of {@code proposedSpec} against {@code currentSpec}. Does not validate the transition. */
+ private boolean areSpecBytesChanged(SupervisorSpec proposedSpec, SupervisorSpec currentSpec)
+ {
+ try {
+ return !Arrays.equals(
+ jsonMapper.writeValueAsBytes(proposedSpec),
+ jsonMapper.writeValueAsBytes(currentSpec)
+ );
+ }
+ catch (JsonProcessingException ex) {
+ log.warn(ex, "Failed to write spec as bytes for spec_id[%s]", proposedSpec.getId());
+ return true;
}
- return true;
}
public boolean stopAndRemoveSupervisor(String id)
@@ -433,7 +469,7 @@ public class SupervisorManager implements SupervisorStatsProvider
Map<String, Object> normalizedEndOffsets = jsonMapper.readValue(jsonMapper.writeValueAsString(endOffsets), Map.class);
BoundedStreamConfig boundedStreamConfig = new BoundedStreamConfig(normalizedStartOffsets, normalizedEndOffsets);
SupervisorSpec backfillSpec = streamSpec.createBackfillSpec(backfillSupervisorId, boundedStreamConfig, backfillTaskCount);
- createOrUpdateAndStartSupervisor(backfillSpec);
+ createOrUpdateAndStartSupervisor(backfillSpec, false);
}
catch (JsonProcessingException e) {
throw new ISE(e, "Failed to serialize offsets for backfill supervisor[%s]", backfillSupervisorId);
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java
index 8d0e04eb798..10297e25a88 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java
@@ -165,29 +165,40 @@ public class SupervisorResource
.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. */
+ private void auditSupervisorUpdate(final SupervisorSpec spec, final HttpServletRequest req)
+ {
+ 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()
+ );
+ }
+
private Set<ResourceAction> getNeededResourceActionsForTask(final SupervisorSpec spec)
{
final Set<ResourceAction> resourceActions =
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
new file mode 100644
index 00000000000..bf14917672c
--- /dev/null
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorSpecUpdateResult.java
@@ -0,0 +1,50 @@
+/*
+ * 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.overlord.supervisor;
+
+/**
+ * Outcome of {@link SupervisorManager#createOrUpdateAndStartSupervisor(SupervisorSpec, boolean)}.
+ */
+public final class SupervisorSpecUpdateResult
+{
+ 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;
+ }
+}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/IdleConfig.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/IdleConfig.java
index 40734815aa6..011ff3c5ca4 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/IdleConfig.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/IdleConfig.java
@@ -24,6 +24,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Preconditions;
import javax.annotation.Nullable;
+import java.util.Objects;
/**
* Defines if and when {@link SeekableStreamSupervisor} can become idle.
@@ -59,6 +60,25 @@ public class IdleConfig
return this.inactiveAfterMillis;
}
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ IdleConfig that = (IdleConfig) o;
+ return enabled == that.enabled && Objects.equals(inactiveAfterMillis, that.inactiveAfterMillis);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(enabled, inactiveAfterMillis);
+ }
+
@Override
public String toString()
{
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/LagAggregator.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/LagAggregator.java
index d84ad70aec9..ed7167cb317 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/LagAggregator.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/LagAggregator.java
@@ -19,6 +19,7 @@
package org.apache.druid.indexing.seekablestream.supervisor;
+import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
import org.apache.druid.indexing.overlord.supervisor.autoscaler.LagStats;
@@ -48,6 +49,28 @@ public interface LagAggregator
*/
class DefaultLagAggregator implements LagAggregator
{
+ private DefaultLagAggregator()
+ {
+ }
+
+ @JsonCreator
+ public static DefaultLagAggregator instance()
+ {
+ return (DefaultLagAggregator) DEFAULT;
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ return o instanceof DefaultLagAggregator;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return DefaultLagAggregator.class.hashCode();
+ }
+
@Override
public <PartitionIdType> LagStats aggregate(Map<PartitionIdType, Long> partitionLags)
{
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfig.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfig.java
index 3c7460d6d59..7b3bd1df76a 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfig.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfig.java
@@ -35,6 +35,7 @@ import org.joda.time.Period;
import javax.annotation.Nullable;
import java.util.Map;
+import java.util.Objects;
public abstract class SeekableStreamSupervisorIOConfig
@@ -316,4 +317,62 @@ public abstract class SeekableStreamSupervisorIOConfig
{
return boundedStreamConfig != null;
}
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SeekableStreamSupervisorIOConfig that = (SeekableStreamSupervisorIOConfig) o;
+ // taskCountExplicit and autoScalerEnabled are construction hints, not part of logical identity.
+ return taskCount == that.taskCount
+ && useEarliestSequenceNumber == that.useEarliestSequenceNumber
+ && Objects.equals(stream, that.stream)
+ && Objects.equals(inputFormat, that.inputFormat)
+ && Objects.equals(replicas, that.replicas)
+ && Objects.equals(taskDuration, that.taskDuration)
+ && Objects.equals(startDelay, that.startDelay)
+ && Objects.equals(period, that.period)
+ && Objects.equals(completionTimeout, that.completionTimeout)
+ && Objects.equals(lateMessageRejectionPeriod, that.lateMessageRejectionPeriod)
+ && Objects.equals(earlyMessageRejectionPeriod, that.earlyMessageRejectionPeriod)
+ && Objects.equals(lateMessageRejectionStartDateTime, that.lateMessageRejectionStartDateTime)
+ && Objects.equals(autoScalerConfig, that.autoScalerConfig)
+ && Objects.equals(idleConfig, that.idleConfig)
+ && Objects.equals(stopTaskCount, that.stopTaskCount)
+ && Objects.equals(serverPriorityToReplicas, that.serverPriorityToReplicas)
+ && Objects.equals(boundedStreamConfig, that.boundedStreamConfig)
+ && Objects.equals(lagAggregator, that.lagAggregator);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(
+ stream,
+ inputFormat,
+ replicas,
+ taskCount,
+ taskDuration,
+ startDelay,
+ period,
+ useEarliestSequenceNumber,
+ completionTimeout,
+ lateMessageRejectionPeriod,
+ earlyMessageRejectionPeriod,
+ lateMessageRejectionStartDateTime,
+ autoScalerConfig,
+ idleConfig,
+ stopTaskCount,
+ serverPriorityToReplicas,
+ boundedStreamConfig,
+ lagAggregator
+ );
+ }
+
+ public abstract SupervisorIOConfigBuilder<?, ?> toBuilder();
}
diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIngestionSpec.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIngestionSpec.java
index 6fbf917b774..dad3dbe2557 100644
--- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIngestionSpec.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIngestionSpec.java
@@ -23,6 +23,8 @@ import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.druid.segment.indexing.DataSchema;
+import java.util.Objects;
+
public abstract class SeekableStreamSupervisorIngestionSpec
{
private final DataSchema dataSchema;
@@ -58,4 +60,25 @@ public abstract class SeekableStreamSupervisorIngestionSpec
{
return tuningConfig;
}
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SeekableStreamSupervisorIngestionSpec that = (SeekableStreamSupervisorIngestionSpec) o;
+ return Objects.equals(getDataSchema(), that.getDataSchema())
+ && Objects.equals(getIOConfig(), that.getIOConfig())
+ && Objects.equals(getTuningConfig(), that.getTuningConfig());
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(getDataSchema(), getIOConfig(), getTuningConfig());
+ }
}
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 ecbd51757c3..6c7da51facb 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
@@ -23,6 +23,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+import org.apache.druid.annotations.SuppressFBWarnings;
import org.apache.druid.common.config.Configs;
import org.apache.druid.error.DruidException;
import org.apache.druid.error.InvalidInput;
@@ -45,6 +46,7 @@ import org.apache.druid.segment.indexing.DataSchema;
import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
public abstract class SeekableStreamSupervisorSpec implements SupervisorSpec
{
@@ -295,6 +297,53 @@ public abstract class SeekableStreamSupervisorSpec implements SupervisorSpec
}
}
+ @Override
+ public boolean requireRestart(SupervisorSpec proposedSpec)
+ {
+ if (!(proposedSpec instanceof SeekableStreamSupervisorSpec proposed)) {
+ return true;
+ }
+
+ final Builder<?> proposedCopy = proposed.toBuilder();
+
+ // When autoscaling is enabled on either side, taskCount is owned by the autoscaler at runtime, so a differing
+ // taskCount does not by itself warrant a stop/recreate. Normalize the proposed copy to the running taskCount so
+ // that only non-taskCount changes can force a restart; the persisted spec still carries the submitted value.
+ if (isAutoScalerEnabled() || proposed.isAutoScalerEnabled()) {
+ proposedCopy.taskCount(getIoConfig().getTaskCount());
+ }
+
+ return !proposedCopy.build().equals(this);
+ }
+
+ private boolean isAutoScalerEnabled()
+ {
+ final AutoScalerConfig autoScalerConfig = getIoConfig().getAutoScalerConfig();
+ return autoScalerConfig != null && autoScalerConfig.getEnableTaskAutoScaler();
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ SeekableStreamSupervisorSpec that = (SeekableStreamSupervisorSpec) o;
+ return suspended == that.suspended
+ && Objects.equals(id, that.id)
+ && Objects.equals(ingestionSchema, that.ingestionSchema)
+ && Objects.equals(context, that.context);
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(id, ingestionSchema, context, suspended);
+ }
+
protected abstract SeekableStreamSupervisorSpec toggleSuspend(boolean suspend);
public abstract SeekableStreamSupervisorSpec createBackfillSpec(
@@ -303,4 +352,85 @@ public abstract class SeekableStreamSupervisorSpec implements SupervisorSpec
@Nullable Integer taskCount
);
+ /**
+ * Returns a copy builder seeded from this spec, used by {@link #requireRestart} for structured comparison.
+ * Abstract so that every stream supervisor spec participates in the restart decision.
+ */
+ public abstract Builder<?> toBuilder();
+
+ @SuppressFBWarnings(
+ value = "URF_UNREAD_PUBLIC_OR_PROTECTED_FIELD",
+ justification = "Fields are populated via copyFrom() and read by build() in concrete subclasses, which "
+ + "live in other modules and so are invisible to SpotBugs' per-module analysis."
+ )
+ public abstract static class Builder<T extends Builder<T>>
+ {
+ protected String id;
+ protected DataSchema dataSchema;
+ protected SeekableStreamSupervisorIOConfig ioConfig;
+ protected SeekableStreamSupervisorTuningConfig tuningConfig;
+ protected Map<String, Object> context;
+ protected Boolean suspended;
+
+ protected abstract T self();
+
+ public abstract SeekableStreamSupervisorSpec build();
+
+ public T copyFrom(SeekableStreamSupervisorSpec spec)
+ {
+ this.id = spec.id;
+ this.dataSchema = spec.getSpec().getDataSchema();
+ this.ioConfig = spec.getIoConfig();
+ this.tuningConfig = spec.getTuningConfig();
+ this.context = spec.context;
+ this.suspended = spec.suspended;
+ return self();
+ }
+
+ public T id(String id)
+ {
+ this.id = id;
+ return self();
+ }
+
+ public T dataSchema(DataSchema dataSchema)
+ {
+ this.dataSchema = dataSchema;
+ return self();
+ }
+
+ public T ioConfig(SeekableStreamSupervisorIOConfig ioConfig)
+ {
+ this.ioConfig = ioConfig;
+ return self();
+ }
+
+ public T tuningConfig(SeekableStreamSupervisorTuningConfig tuningConfig)
+ {
+ this.tuningConfig = tuningConfig;
+ return self();
+ }
+
+ public T context(Map<String, Object> context)
+ {
+ this.context = context;
+ return self();
+ }
+
+ public T suspended(boolean suspended)
+ {
+ this.suspended = suspended;
+ return self();
+ }
+
+ /**
+ * Sets {@code ioConfig.taskCount} on a copy (does not mutate the builder's current ioConfig reference).
+ */
+ public T taskCount(int taskCount)
+ {
+ this.ioConfig = this.ioConfig.toBuilder().withTaskCount(taskCount).build();
+ return self();
+ }
+ }
+
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SupervisorIOConfigBuilder.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SupervisorIOConfigBuilder.java
similarity index 53%
rename from indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SupervisorIOConfigBuilder.java
rename to indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SupervisorIOConfigBuilder.java
index bcfb33e9d12..d0ebe35a1fb 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SupervisorIOConfigBuilder.java
+++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SupervisorIOConfigBuilder.java
@@ -23,10 +23,15 @@ import org.apache.druid.data.input.InputFormat;
import org.apache.druid.data.input.impl.JsonInputFormat;
import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.AutoScalerConfig;
import org.joda.time.DateTime;
+import org.joda.time.Duration;
import org.joda.time.Period;
+import java.util.Map;
+
/**
- * Builder for a {@link SeekableStreamSupervisorIOConfig}.
+ * Builder for a {@link SeekableStreamSupervisorIOConfig}. Subclasses add stream-specific fields and
+ * implement {@link #build()}. {@link #copyFromBase(SeekableStreamSupervisorIOConfig)} populates the
+ * base-class fields from an existing config, enabling {@code toBuilder()}-style copies.
*
* @param <Self> Type of this builder itself
* @param <C> Type of the IOConfig created by this builder
@@ -51,6 +56,8 @@ public abstract class SupervisorIOConfigBuilder<
protected DateTime lateMessageRejectionStartDateTime;
protected IdleConfig idleConfig;
protected Integer stopTaskCount;
+ protected Map<Integer, Integer> serverPriorityToReplicas;
+ protected BoundedStreamConfig boundedStreamConfig;
public Self withStream(String stream)
{
@@ -154,11 +161,99 @@ public abstract class SupervisorIOConfigBuilder<
return self();
}
+ public Self withServerPriorityToReplicas(Map<Integer, Integer> serverPriorityToReplicas)
+ {
+ this.serverPriorityToReplicas = serverPriorityToReplicas;
+ return self();
+ }
+
+ public Self withBoundedStreamConfig(BoundedStreamConfig boundedStreamConfig)
+ {
+ this.boundedStreamConfig = boundedStreamConfig;
+ return self();
+ }
+
+ /**
+ * Populates this builder's base-class fields from an existing config. Stores the constructor's
+ * input types ({@link Period} rather than the normalized {@link Duration}) so {@link #build()}
+ * round-trips through the public constructor.
+ */
+ public Self copyFromBase(SeekableStreamSupervisorIOConfig io)
+ {
+ this.stream = io.getStream();
+ this.inputFormat = io.getInputFormat();
+ this.replicas = io.getReplicas();
+ this.taskCount = io.getTaskCount();
+ this.taskDuration = toPeriod(io.getTaskDuration());
+ this.startDelay = toPeriod(io.getStartDelay());
+ this.period = toPeriod(io.getPeriod());
+ this.useEarliestSequenceNumber = io.isUseEarliestSequenceNumber();
+ this.completionTimeout = toPeriod(io.getCompletionTimeout());
+ this.lateMessageRejectionPeriod = io.getLateMessageRejectionPeriod().isPresent()
+ ? toPeriod(io.getLateMessageRejectionPeriod().get()) : null;
+ this.earlyMessageRejectionPeriod = io.getEarlyMessageRejectionPeriod().isPresent()
+ ? toPeriod(io.getEarlyMessageRejectionPeriod().get()) : null;
+ this.lateMessageRejectionStartDateTime = io.getLateMessageRejectionStartDateTime().orNull();
+ this.autoScalerConfig = io.getAutoScalerConfig();
+ this.lagAggregator = io.getLagAggregator();
+ this.idleConfig = io.getIdleConfig();
+ this.stopTaskCount = io.getStopTaskCount();
+ this.serverPriorityToReplicas = io.getServerPriorityToReplicas();
+ this.boundedStreamConfig = io.getBoundedStreamConfig();
+ return self();
+ }
+
public abstract C build();
+ protected static Period toPeriod(Duration duration)
+ {
+ return duration == null ? null : new Period(duration);
+ }
+
@SuppressWarnings("unchecked")
- private Self self()
+ protected Self self()
{
return (Self) this;
}
+
+ /**
+ * Concrete builder producing a plain {@link SeekableStreamSupervisorIOConfig}. All instances are of
+ * the single anonymous class defined in {@link #build()}, so builder-produced configs share a
+ * runtime class and compare correctly via {@link SeekableStreamSupervisorIOConfig#equals(Object)}.
+ */
+ public static final class DefaultSupervisorIOConfigBuilder
+ extends SupervisorIOConfigBuilder<DefaultSupervisorIOConfigBuilder, SeekableStreamSupervisorIOConfig>
+ {
+ @Override
+ public SeekableStreamSupervisorIOConfig build()
+ {
+ return new SeekableStreamSupervisorIOConfig(
+ stream,
+ inputFormat,
+ replicas,
+ taskCount,
+ taskDuration,
+ startDelay,
+ period,
+ useEarliestSequenceNumber,
+ completionTimeout,
+ lateMessageRejectionPeriod,
+ earlyMessageRejectionPeriod,
+ autoScalerConfig,
+ lagAggregator,
+ lateMessageRejectionStartDateTime,
+ idleConfig,
+ stopTaskCount,
+ serverPriorityToReplicas,
+ boundedStreamConfig
+ )
+ {
+ @Override
+ public SupervisorIOConfigBuilder<?, ?> toBuilder()
+ {
+ return new DefaultSupervisorIOConfigBuilder().copyFromBase(this);
+ }
+ };
+ }
+ }
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentTransactionalInsertActionTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentTransactionalInsertActionTest.java
index 217aae3f258..49428928a56 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentTransactionalInsertActionTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/actions/SegmentTransactionalInsertActionTest.java
@@ -194,7 +194,7 @@ public class SegmentTransactionalInsertActionTest
acquireTimeChunkLock(TaskLockType.EXCLUSIVE, task, INTERVAL, 5000);
final NoopSupervisorSpec supervisorSpec = new NoopSupervisorSpec(SUPERVISOR_ID, List.of(DATA_SOURCE));
- actionTestKit.getSupervisorManager().createOrUpdateAndStartSupervisor(supervisorSpec);
+ actionTestKit.getSupervisorManager().createOrUpdateAndStartSupervisor(supervisorSpec, false);
SegmentPublishResult result = SegmentTransactionalInsertAction.appendAction(
ImmutableSet.of(SEGMENT1),
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 7872ffa4f2d..940dd211542 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.SupervisorSpecUpdateResult;
import org.apache.druid.indexing.overlord.supervisor.VersionedSupervisorSpec;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.segment.TestDataSource;
@@ -315,10 +316,8 @@ public class OverlordCompactionResourceTest
final CompactionSupervisorSpec supervisorSpec =
new CompactionSupervisorSpec(wikiConfig, false, validator);
- EasyMock.expect(supervisorManager.shouldUpdateSupervisor(supervisorSpec))
- .andReturn(true).once();
- EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(supervisorSpec))
- .andReturn(true).once();
+ EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(supervisorSpec, true))
+ .andReturn(SupervisorSpecUpdateResult.of(true, true)).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 23e9f94fa8b..ec104eb2215 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
@@ -46,6 +46,7 @@ import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervi
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.SupervisorIOConfigBuilder;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.Intervals;
@@ -128,7 +129,7 @@ public class SupervisorManagerTest extends EasyMockSupport
manager.start();
Assert.assertEquals(1, manager.getSupervisorIds().size());
- manager.createOrUpdateAndStartSupervisor(spec);
+ manager.createOrUpdateAndStartSupervisor(spec, false);
Assert.assertEquals(2, manager.getSupervisorIds().size());
Assert.assertEquals(spec, manager.getSupervisorSpec("id1").get());
verifyAll();
@@ -139,7 +140,7 @@ public class SupervisorManagerTest extends EasyMockSupport
supervisor1.stop(true);
replayAll();
- manager.createOrUpdateAndStartSupervisor(spec2);
+ manager.createOrUpdateAndStartSupervisor(spec2, false);
Assert.assertEquals(2, manager.getSupervisorIds().size());
Assert.assertEquals(spec2, manager.getSupervisorSpec("id1").get());
verifyAll();
@@ -191,7 +192,7 @@ public class SupervisorManagerTest extends EasyMockSupport
manager.start();
Assert.assertEquals(0, manager.getSupervisorIds().size());
- manager.createOrUpdateAndStartSupervisor(spec);
+ manager.createOrUpdateAndStartSupervisor(spec, false);
Assert.assertEquals(1, manager.getSupervisorIds().size());
Assert.assertEquals(spec, manager.getSupervisorSpec("id1").get());
verifyAll();
@@ -200,7 +201,7 @@ public class SupervisorManagerTest extends EasyMockSupport
exception.expect(DruidException.class);
replayAll();
- manager.createOrUpdateAndStartSupervisor(spec2);
+ manager.createOrUpdateAndStartSupervisor(spec2, false);
verifyAll();
}
@@ -208,7 +209,7 @@ public class SupervisorManagerTest extends EasyMockSupport
public void testCreateOrUpdateAndStartSupervisorNotStarted()
{
exception.expect(IllegalStateException.class);
- manager.createOrUpdateAndStartSupervisor(new TestSupervisorSpec("id", null));
+ manager.createOrUpdateAndStartSupervisor(new TestSupervisorSpec("id", null), false);
}
@Test
@@ -220,7 +221,7 @@ public class SupervisorManagerTest extends EasyMockSupport
exception.expect(NullPointerException.class);
manager.start();
- manager.createOrUpdateAndStartSupervisor(null);
+ manager.createOrUpdateAndStartSupervisor(null, false);
verifyAll();
}
@@ -233,26 +234,167 @@ public class SupervisorManagerTest extends EasyMockSupport
exception.expect(NullPointerException.class);
manager.start();
- manager.createOrUpdateAndStartSupervisor(new TestSupervisorSpec(null, null));
+ manager.createOrUpdateAndStartSupervisor(new TestSupervisorSpec(null, null), false);
verifyAll();
}
@Test
- public void testShouldUpdateSupervisor()
+ public void testUnchangedSpecDoesNotRestart()
{
SupervisorSpec spec = new TestSupervisorSpec("id1", supervisor1);
+ Map<String, SupervisorSpec> existingSpecs = Map.of("id1", spec);
+ EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(existingSpecs);
+ supervisor1.start();
+ EasyMock.expect(supervisor1.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
+ replayAll();
+ manager.start();
+ final SupervisorSpecUpdateResult result = manager.createOrUpdateAndStartSupervisor(spec, true);
+ Assert.assertFalse(result.isModified());
+ Assert.assertFalse(result.isRestarted());
+ verifyAll();
+ }
+
+ @Test
+ public void testNewSupervisorRequiresRestart()
+ {
SupervisorSpec spec2 = new TestSupervisorSpec("id2", supervisor2);
Map<String, SupervisorSpec> existingSpecs = ImmutableMap.of(
- "id1", spec
+ "id1", new TestSupervisorSpec("id1", supervisor1)
);
EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(existingSpecs);
supervisor1.start();
EasyMock.expect(supervisor1.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
replayAll();
manager.start();
- Assert.assertFalse(manager.shouldUpdateSupervisor(spec));
- Assert.assertTrue(manager.shouldUpdateSupervisor(spec2));
- Assert.assertTrue(manager.shouldUpdateSupervisor(new NoopSupervisorSpec("id1", null)));
+ verifyAll();
+
+ resetAll();
+ supervisor2.start();
+ EasyMock.expect(supervisor2.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
+ metadataSupervisorManager.insert(EasyMock.eq("id2"), EasyMock.anyObject());
+ replayAll();
+ final SupervisorSpecUpdateResult result = manager.createOrUpdateAndStartSupervisor(spec2, true);
+ Assert.assertTrue(result.isModified());
+ Assert.assertTrue(result.isRestarted());
+ verifyAll();
+ }
+
+ @Test
+ public void testCreateOrUpdateSkipRestart_changedNoRestart_persistsWithoutRestart()
+ {
+ final Capture<TestSupervisorSpec> capturedInsert = Capture.newInstance();
+ final Map<String, SupervisorSpec> existingSpecs =
+ Map.of("id1", new VersionedTestSupervisorSpec("id1", supervisor1, 1));
+ EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(existingSpecs);
+ supervisor1.start();
+ EasyMock.expect(supervisor1.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
+ // Persist is expected; supervisor1.stop(...) is NOT expected, so verifyAll() asserts no restart.
+ metadataSupervisorManager.insert(EasyMock.eq("id1"), EasyMock.capture(capturedInsert));
+ replayAll();
+
+ manager.start();
+ // version differs (byte change) but requireRestart() is false.
+ final SupervisorSpec updated = new VersionedTestSupervisorSpec("id1", supervisor1, 2);
+ final SupervisorSpecUpdateResult updateResult = manager.createOrUpdateAndStartSupervisor(updated, true);
+ Assert.assertTrue(updateResult.isModified());
+ Assert.assertFalse(updateResult.isRestarted());
+ Assert.assertSame(updated, capturedInsert.getValue());
+ Assert.assertSame(updated, manager.getSupervisorSpec("id1").get());
+ verifyAll();
+ }
+
+ @Test
+ public void testCreateOrUpdateSkipRestart_identicalSpec_isNoop()
+ {
+ final Map<String, SupervisorSpec> existingSpecs = Map.of("id1", new TestSupervisorSpec("id1", supervisor1));
+ EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(existingSpecs);
+ supervisor1.start();
+ EasyMock.expect(supervisor1.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
+ // No insert and no stop expected for an identical spec.
+ replayAll();
+
+ manager.start();
+ final SupervisorSpecUpdateResult updateResult =
+ manager.createOrUpdateAndStartSupervisor(new TestSupervisorSpec("id1", supervisor1), true);
+ Assert.assertFalse(updateResult.isModified());
+ Assert.assertFalse(updateResult.isRestarted());
+ verifyAll();
+ }
+
+ @Test
+ public void testCreateOrUpdateSkipRestart_changedRequiringRestart_restartsAndPersists()
+ {
+ // The restart decision belongs to the running spec, so requireRestart=true is set on the existing spec.
+ final Map<String, SupervisorSpec> existingSpecs =
+ Map.of("id1", new VersionedTestSupervisorSpec("id1", supervisor1, 1, true));
+ EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(existingSpecs);
+ supervisor1.start();
+ EasyMock.expect(supervisor1.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
+ replayAll();
+ manager.start();
+ verifyAll();
+
+ resetAll();
+ // A changed spec that requires a restart must be fully applied (stop + recreate + persist), even on the
+ // skip-restart path — it must never be silently dropped or persisted without restarting.
+ supervisor1.stop(true);
+ supervisor1.start();
+ EasyMock.expect(supervisor1.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
+ metadataSupervisorManager.insert(EasyMock.eq("id1"), EasyMock.anyObject());
+ replayAll();
+
+ final SupervisorSpec updated = new VersionedTestSupervisorSpec("id1", supervisor1, 2);
+ final SupervisorSpecUpdateResult updateResult = manager.createOrUpdateAndStartSupervisor(updated, true);
+ Assert.assertTrue(updateResult.isModified());
+ Assert.assertTrue(updateResult.isRestarted());
+ Assert.assertSame(updated, manager.getSupervisorSpec("id1").get());
+ verifyAll();
+ }
+
+ @Test
+ public void testCreateOrUpdateSkipRestart_mergesBeforeDiffAndRestartDecision()
+ {
+ final MergingVersionedTestSupervisorSpec existing =
+ new MergingVersionedTestSupervisorSpec("id1", supervisor1, 1);
+ final Map<String, SupervisorSpec> existingSpecs = Map.of("id1", existing);
+ EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(existingSpecs);
+ supervisor1.start();
+ EasyMock.expect(supervisor1.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
+ replayAll();
+
+ manager.start();
+ final MergingVersionedTestSupervisorSpec updated =
+ new MergingVersionedTestSupervisorSpec("id1", supervisor1, null);
+ final SupervisorSpecUpdateResult updateResult = manager.createOrUpdateAndStartSupervisor(updated, true);
+ Assert.assertFalse(updateResult.isModified());
+ Assert.assertFalse(updateResult.isRestarted());
+ Assert.assertEquals(Integer.valueOf(1), updated.getVersion());
+ Assert.assertSame(existing, manager.getSupervisorSpec("id1").get());
+ verifyAll();
+ }
+
+ @Test
+ public void testCreateOrUpdateWithoutSkipRestart_forcesRestartWhenSpecUnchanged()
+ {
+ final Map<String, SupervisorSpec> existingSpecs = Map.of("id1", new TestSupervisorSpec("id1", supervisor1));
+ EasyMock.expect(metadataSupervisorManager.getLatest()).andReturn(existingSpecs);
+ supervisor1.start();
+ EasyMock.expect(supervisor1.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
+ replayAll();
+ manager.start();
+ verifyAll();
+
+ resetAll();
+ supervisor1.stop(true);
+ supervisor1.start();
+ EasyMock.expect(supervisor1.createAutoscaler(EasyMock.anyObject())).andReturn(null).anyTimes();
+ replayAll();
+
+ final SupervisorSpecUpdateResult updateResult =
+ manager.createOrUpdateAndStartSupervisor(new TestSupervisorSpec("id1", supervisor1), false);
+ Assert.assertFalse(updateResult.isModified());
+ Assert.assertTrue(updateResult.isRestarted());
+ verifyAll();
}
@Test
@@ -276,7 +418,7 @@ public class SupervisorManagerTest extends EasyMockSupport
exception.expect(DruidException.class);
replayAll();
manager.start();
- manager.shouldUpdateSupervisor(spec2);
+ manager.createOrUpdateAndStartSupervisor(spec2, true);
verifyAll();
}
@@ -592,8 +734,8 @@ public class SupervisorManagerTest extends EasyMockSupport
manager.start();
try {
- manager.createOrUpdateAndStartSupervisor(testSpecOld);
- manager.createOrUpdateAndStartSupervisor(testSpecNew);
+ manager.createOrUpdateAndStartSupervisor(testSpecOld, false);
+ manager.createOrUpdateAndStartSupervisor(testSpecNew, false);
}
catch (Exception e) {
Assert.assertEquals(testSpecOld, capturedInsert.getValue());
@@ -719,7 +861,7 @@ public class SupervisorManagerTest extends EasyMockSupport
manager.start();
Assert.assertEquals(1, manager.getSupervisorIds().size());
- manager.createOrUpdateAndStartSupervisor(spec);
+ manager.createOrUpdateAndStartSupervisor(spec, false);
Assert.assertEquals(2, manager.getSupervisorIds().size());
Assert.assertEquals(spec, manager.getSupervisorSpec("id1").get());
verifyAll();
@@ -862,22 +1004,22 @@ public class SupervisorManagerTest extends EasyMockSupport
Assert.assertTrue(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("nonExistent").isEmpty());
- manager.createOrUpdateAndStartSupervisor(noopSupervisorSpec);
+ manager.createOrUpdateAndStartSupervisor(noopSupervisorSpec, false);
Assert.assertTrue(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("noopDS").isEmpty());
- manager.createOrUpdateAndStartSupervisor(suspendedSpec);
- Assert.assertTrue(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("suspendedDS").isEmpty());
+ manager.createOrUpdateAndStartSupervisor(suspendedSpec, false);
+ Assert.assertTrue(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("activeDS").isEmpty());
- manager.createOrUpdateAndStartSupervisor(activeSpec);
+ manager.createOrUpdateAndStartSupervisor(activeSpec, false);
Assert.assertTrue(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("activeDS").isEmpty());
- manager.createOrUpdateAndStartSupervisor(activeAppendSpec);
+ manager.createOrUpdateAndStartSupervisor(activeAppendSpec, false);
Assert.assertFalse(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("activeAppendDS").isEmpty());
- manager.createOrUpdateAndStartSupervisor(activeSpecWithConcurrentLocks);
+ manager.createOrUpdateAndStartSupervisor(activeSpecWithConcurrentLocks, false);
Assert.assertFalse(manager.getActiveSupervisorIdsForDatasourceWithAppendLock("activeConcurrentLocksDS").isEmpty());
- manager.createOrUpdateAndStartSupervisor(specWithUseConcurrentLocksFalse);
+ manager.createOrUpdateAndStartSupervisor(specWithUseConcurrentLocksFalse, false);
Assert.assertTrue(
manager.getActiveSupervisorIdsForDatasourceWithAppendLock("dsWithUseConcurrentLocksFalse").isEmpty()
);
@@ -922,10 +1064,10 @@ public class SupervisorManagerTest extends EasyMockSupport
);
manager.start();
- manager.createOrUpdateAndStartSupervisor(noopSpec);
+ manager.createOrUpdateAndStartSupervisor(noopSpec, false);
Assert.assertFalse(manager.registerUpgradedPendingSegmentOnSupervisor("noop", pendingSegment).isPresent());
- manager.createOrUpdateAndStartSupervisor(streamingSpec);
+ manager.createOrUpdateAndStartSupervisor(streamingSpec, false);
Assert.assertEquals(OptionalInt.of(1), manager.registerUpgradedPendingSegmentOnSupervisor("sss", pendingSegment));
verifyAll();
@@ -1392,6 +1534,12 @@ public class SupervisorManagerTest extends EasyMockSupport
return getSpec().getIOConfig();
}
+ @Override
+ public Builder<?> toBuilder()
+ {
+ throw new UnsupportedOperationException();
+ }
+
@JsonTypeName("testBackfillIngestionSpec")
static class IngestionSpec extends SeekableStreamSupervisorIngestionSpec
{
@@ -1424,6 +1572,83 @@ public class SupervisorManagerTest extends EasyMockSupport
{
super(stream, null, 1, taskCount, null, null, null, false, null, null, null, null, LagAggregator.DEFAULT, null, null, null, null, boundedStreamConfig);
}
+
+ @Override
+ public SupervisorIOConfigBuilder<?, ?> toBuilder()
+ {
+ return new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder().copyFromBase(this);
+ }
+ }
+ }
+
+ /**
+ * A {@link TestSupervisorSpec} with an explicitly-serialized {@code version} field, so two specs
+ * sharing the same id can still differ in their serialized form (the base spec serializes only
+ * id-derived properties).
+ */
+ private static class VersionedTestSupervisorSpec extends TestSupervisorSpec
+ {
+ private final int version;
+ private final boolean requireRestart;
+
+ VersionedTestSupervisorSpec(String id, Supervisor supervisor, int version)
+ {
+ this(id, supervisor, version, false);
+ }
+
+ VersionedTestSupervisorSpec(String id, Supervisor supervisor, int version, boolean requireRestart)
+ {
+ super(id, supervisor);
+ this.version = version;
+ this.requireRestart = requireRestart;
+ }
+
+ @JsonProperty
+ public int getVersion()
+ {
+ return version;
+ }
+
+ /**
+ * 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.)
+ */
+ @Override
+ public boolean requireRestart(SupervisorSpec proposedSpec)
+ {
+ return requireRestart;
+ }
+ }
+
+ private static class MergingVersionedTestSupervisorSpec extends TestSupervisorSpec
+ {
+ @Nullable
+ private Integer version;
+
+ MergingVersionedTestSupervisorSpec(
+ final String id,
+ final Supervisor supervisor,
+ @Nullable final Integer version
+ )
+ {
+ super(id, supervisor);
+ this.version = version;
+ }
+
+ @JsonProperty
+ @Nullable
+ public Integer getVersion()
+ {
+ return version;
+ }
+
+ @Override
+ public void merge(@Nullable final SupervisorSpec existingSpec)
+ {
+ if (version == null && existingSpec instanceof MergingVersionedTestSupervisorSpec) {
+ version = ((MergingVersionedTestSupervisorSpec) existingSpec).version;
+ }
}
}
}
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 31e0d604a22..1ed1214368b 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
@@ -164,7 +164,8 @@ public class SupervisorResourceTest extends EasyMockSupport
};
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
- EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec)).andReturn(true);
+ EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec, false))
+ .andReturn(SupervisorSpecUpdateResult.of(true, true));
setupMockRequest();
setupMockRequestForAudit();
@@ -179,7 +180,27 @@ public class SupervisorResourceTest extends EasyMockSupport
verifyAll();
Assert.assertEquals(200, response.getStatus());
- Assert.assertEquals(ImmutableMap.of("id", "my-id", "restarted", true), response.getEntity());
+ Assert.assertEquals(Map.of("id", "my-id", "modified", true, "restarted", true), response.getEntity());
+ resetAll();
+
+ EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
+ EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec, false))
+ .andReturn(SupervisorSpecUpdateResult.of(false, true));
+
+ setupMockRequest();
+ setupMockRequestForAudit();
+
+ EasyMock.expect(authConfig.isEnableInputSourceSecurity()).andReturn(true);
+ auditManager.doAudit(EasyMock.anyObject());
+ EasyMock.expectLastCall().once();
+
+ replayAll();
+
+ response = supervisorResource.specPost(spec, false, request);
+ verifyAll();
+
+ Assert.assertEquals(200, response.getStatus());
+ Assert.assertEquals(Map.of("id", "my-id", "modified", false, "restarted", true), response.getEntity());
resetAll();
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent());
@@ -238,24 +259,47 @@ public class SupervisorResourceTest extends EasyMockSupport
};
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
- EasyMock.expect(supervisorManager.shouldUpdateSupervisor(spec)).andReturn(false);
+ // 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));
setupMockRequest();
+ setupMockRequestForAudit();
EasyMock.expect(authConfig.isEnableInputSourceSecurity()).andReturn(true);
+ auditManager.doAudit(EasyMock.anyObject());
+ EasyMock.expectLastCall().once();
replayAll();
Response response = supervisorResource.specPost(spec, true, request);
verifyAll();
Assert.assertEquals(200, response.getStatus());
- Assert.assertEquals(ImmutableMap.of("id", "my-id", "restarted", false), response.getEntity());
+ Assert.assertEquals(Map.of("id", "my-id", "modified", true, "restarted", false), response.getEntity());
resetAll();
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
- EasyMock.expect(supervisorManager.shouldUpdateSupervisor(spec)).andReturn(true);
- EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec)).andReturn(true);
+ EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec, true))
+ .andReturn(SupervisorSpecUpdateResult.of(false, false));
+
+ setupMockRequest();
+
+ EasyMock.expect(authConfig.isEnableInputSourceSecurity()).andReturn(true);
+
+ replayAll();
+
+ response = supervisorResource.specPost(spec, true, request);
+ verifyAll();
+
+ Assert.assertEquals(200, response.getStatus());
+ Assert.assertEquals(Map.of("id", "my-id", "modified", false, "restarted", false), response.getEntity());
+
+ resetAll();
+
+ EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
+ EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec, true))
+ .andReturn(SupervisorSpecUpdateResult.of(true, true));
setupMockRequest();
setupMockRequestForAudit();
@@ -270,7 +314,7 @@ public class SupervisorResourceTest extends EasyMockSupport
verifyAll();
Assert.assertEquals(200, response.getStatus());
- Assert.assertEquals(ImmutableMap.of("id", "my-id", "restarted", true), response.getEntity());
+ Assert.assertEquals(Map.of("id", "my-id", "modified", true, "restarted", true), response.getEntity());
}
@Test
@@ -287,7 +331,8 @@ public class SupervisorResourceTest extends EasyMockSupport
};
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.of(supervisorManager));
- EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec)).andReturn(true);
+ EasyMock.expect(supervisorManager.createOrUpdateAndStartSupervisor(spec, false))
+ .andReturn(SupervisorSpecUpdateResult.of(true, true));
setupMockRequest();
setupMockRequestForAudit();
@@ -301,7 +346,7 @@ public class SupervisorResourceTest extends EasyMockSupport
verifyAll();
Assert.assertEquals(200, response.getStatus());
- Assert.assertEquals(ImmutableMap.of("id", "my-id", "restarted", true), response.getEntity());
+ Assert.assertEquals(Map.of("id", "my-id", "modified", true, "restarted", true), response.getEntity());
resetAll();
EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent());
@@ -1773,6 +1818,12 @@ public class SupervisorResourceTest extends EasyMockSupport
return null;
}
+ @Override
+ public Builder<?> toBuilder()
+ {
+ throw new UnsupportedOperationException();
+ }
+
@JsonIgnore
@Nonnull
@Override
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamSamplerSpecTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamSamplerSpecTest.java
index 6fd6307e50c..4a5ba773e89 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamSamplerSpecTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamSamplerSpecTest.java
@@ -42,6 +42,7 @@ import org.apache.druid.indexing.seekablestream.supervisor.IdleConfig;
import org.apache.druid.indexing.seekablestream.supervisor.LagAggregator;
import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorIOConfig;
import org.apache.druid.indexing.seekablestream.supervisor.SeekableStreamSupervisorSpec;
+import org.apache.druid.indexing.seekablestream.supervisor.SupervisorIOConfigBuilder;
import org.apache.druid.indexing.seekablestream.supervisor.autoscaler.AutoScalerConfig;
import org.apache.druid.jackson.DefaultObjectMapper;
import org.apache.druid.java.util.common.StringUtils;
@@ -337,5 +338,11 @@ public class SeekableStreamSamplerSpecTest extends EasyMockSupport
null
);
}
+
+ @Override
+ public SupervisorIOConfigBuilder<?, ?> toBuilder()
+ {
+ return new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder().copyFromBase(this);
+ }
}
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfigTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfigTest.java
index ec60d5753f7..d9837e784e8 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfigTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorIOConfigTest.java
@@ -19,6 +19,8 @@
package org.apache.druid.indexing.seekablestream.supervisor;
+import nl.jqno.equalsverifier.EqualsVerifier;
+import nl.jqno.equalsverifier.Warning;
import org.apache.druid.data.input.InputFormat;
import org.apache.druid.error.DruidException;
import org.apache.druid.error.DruidExceptionMatcher;
@@ -27,6 +29,7 @@ import org.apache.druid.java.util.common.DateTimes;
import org.apache.druid.java.util.common.IAE;
import org.apache.druid.java.util.common.StringUtils;
import org.hamcrest.MatcherAssert;
+import org.joda.time.DateTime;
import org.joda.time.Duration;
import org.joda.time.Period;
import org.junit.Assert;
@@ -52,7 +55,7 @@ public class SeekableStreamSupervisorIOConfigTest
LagAggregator lagAggregator = mock(LagAggregator.class);
InputFormat inputFormat = mock(InputFormat.class);
- SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
+ SeekableStreamSupervisorIOConfig config = new TestableSeekableStreamSupervisorIOConfig(
"stream",
inputFormat,
null,
@@ -129,7 +132,7 @@ public class SeekableStreamSupervisorIOConfigTest
boolean expectedExplicit
)
{
- final SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
+ final SeekableStreamSupervisorIOConfig config = new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
2,
@@ -162,7 +165,7 @@ public class SeekableStreamSupervisorIOConfigTest
IAE ex = Assert.assertThrows(
IAE.class,
- () -> new SeekableStreamSupervisorIOConfig(
+ () -> new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -198,7 +201,7 @@ public class SeekableStreamSupervisorIOConfigTest
{
DruidException ex = Assert.assertThrows(
DruidException.class,
- () -> new SeekableStreamSupervisorIOConfig(
+ () -> new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -232,7 +235,7 @@ public class SeekableStreamSupervisorIOConfigTest
LagAggregator lagAggregator = mock(LagAggregator.class);
// Autoscaler disabled, stopTaskCount unset
- SeekableStreamSupervisorIOConfig config1 = new SeekableStreamSupervisorIOConfig(
+ SeekableStreamSupervisorIOConfig config1 = new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -257,7 +260,7 @@ public class SeekableStreamSupervisorIOConfigTest
Assert.assertEquals(7, config1.getMaxAllowedStops());
// Autoscaler disabled, stopTaskCount set
- SeekableStreamSupervisorIOConfig config2 = new SeekableStreamSupervisorIOConfig(
+ SeekableStreamSupervisorIOConfig config2 = new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -294,7 +297,7 @@ public class SeekableStreamSupervisorIOConfigTest
when(autoScalerConfig.getTaskCountStart()).thenReturn(10);
when(autoScalerConfig.getStopTaskCountRatio()).thenReturn(0.5);
- SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
+ SeekableStreamSupervisorIOConfig config = new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -328,7 +331,7 @@ public class SeekableStreamSupervisorIOConfigTest
when(autoScalerConfig.getTaskCountStart()).thenReturn(10);
when(autoScalerConfig.getStopTaskCountRatio()).thenReturn(null);
- SeekableStreamSupervisorIOConfig config2 = new SeekableStreamSupervisorIOConfig(
+ SeekableStreamSupervisorIOConfig config2 = new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -359,7 +362,7 @@ public class SeekableStreamSupervisorIOConfigTest
when(autoScalerConfig.getTaskCountStart()).thenReturn(10);
when(autoScalerConfig.getStopTaskCountRatio()).thenReturn(null);
- SeekableStreamSupervisorIOConfig config3 = new SeekableStreamSupervisorIOConfig(
+ SeekableStreamSupervisorIOConfig config3 = new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -447,7 +450,7 @@ public class SeekableStreamSupervisorIOConfigTest
private SeekableStreamSupervisorIOConfig makeSeekableStreamSupervisorIOConfig(@Nullable Integer replicas, @Nullable Map<Integer, Integer> serverPriorityToReplicas)
{
- return new SeekableStreamSupervisorIOConfig(
+ return new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
replicas,
@@ -480,7 +483,7 @@ public class SeekableStreamSupervisorIOConfigTest
LagAggregator lagAggregator = mock(LagAggregator.class);
- SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
+ SeekableStreamSupervisorIOConfig config = new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -513,7 +516,7 @@ public class SeekableStreamSupervisorIOConfigTest
{
LagAggregator lagAggregator = mock(LagAggregator.class);
- SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
+ SeekableStreamSupervisorIOConfig config = new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -545,7 +548,7 @@ public class SeekableStreamSupervisorIOConfigTest
{
LagAggregator lagAggregator = mock(LagAggregator.class);
- SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
+ SeekableStreamSupervisorIOConfig config = new TestableSeekableStreamSupervisorIOConfig(
"stream",
null,
null,
@@ -571,4 +574,122 @@ public class SeekableStreamSupervisorIOConfigTest
Assert.assertFalse(config.isBounded());
Assert.assertNull(config.getBoundedStreamConfig());
}
+
+ private static SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder ioConfigBuilder()
+ {
+ return new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withReplicas(1)
+ .withTaskCount(2)
+ .withTaskDuration(new Period("PT1H"))
+ .withLagAggregator(LagAggregator.DEFAULT);
+ }
+
+ @Test
+ public void testEqualsAndHashCode()
+ {
+ final SeekableStreamSupervisorIOConfig config = ioConfigBuilder().build();
+ Assert.assertEquals(config, ioConfigBuilder().build());
+ Assert.assertEquals(config.hashCode(), ioConfigBuilder().build().hashCode());
+ Assert.assertNotEquals(config, null);
+ Assert.assertNotEquals(config, "not an io config");
+ Assert.assertNotEquals(config, ioConfigBuilder().withStream("other").build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withReplicas(9).build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withTaskCount(9).build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withStopTaskCount(7).build());
+ Assert.assertNotEquals(config, ioConfigBuilder().withIdleConfig(new IdleConfig(true, 5L)).build());
+ }
+
+ @Test
+ public void testIdleConfigEqualsAndHashCode()
+ {
+ EqualsVerifier.forClass(IdleConfig.class).usingGetClass().verify();
+ }
+
+ /**
+ * Drift guard: the supervisor restart decision is equality-based, so any field omitted from
+ * {@code equals} would let a changed spec persist without restarting. EqualsVerifier reflects over the
+ * fields and fails automatically on a newly-added unused field — only abstract field types need prefab
+ * values. {@code taskCountExplicit}/{@code autoScalerEnabled} are derived hints (ignored); {@code taskCount}
+ * is mutable (NONFINAL_FIELDS suppressed).
+ */
+ @Test
+ public void testEqualsContractCoversAllFields()
+ {
+ EqualsVerifier.forClass(SeekableStreamSupervisorIOConfig.class)
+ .usingGetClass()
+ .withIgnoredFields("taskCountExplicit", "autoScalerEnabled")
+ .suppress(Warning.NONFINAL_FIELDS)
+ .withPrefabValues(InputFormat.class, mock(InputFormat.class), mock(InputFormat.class))
+ .withPrefabValues(AutoScalerConfig.class, mock(AutoScalerConfig.class), mock(AutoScalerConfig.class))
+ .withPrefabValues(LagAggregator.class, mock(LagAggregator.class), mock(LagAggregator.class))
+ .verify();
+ }
+
+ @Test
+ public void testDefaultLagAggregatorEquals()
+ {
+ // The default aggregator is a stateless singleton; instance() always returns DEFAULT.
+ final LagAggregator aggregator = LagAggregator.DefaultLagAggregator.instance();
+ Assert.assertSame(LagAggregator.DEFAULT, aggregator);
+ Assert.assertEquals(aggregator, LagAggregator.DefaultLagAggregator.instance());
+ Assert.assertNotEquals(aggregator, null);
+ Assert.assertNotEquals(aggregator, "not a lag aggregator");
+ }
+
+ /**
+ * Concrete subclass for tests that exercise the abstract base directly. Implements the now-abstract
+ * {@link SeekableStreamSupervisorIOConfig#toBuilder()} via the generic builder.
+ */
+ private static class TestableSeekableStreamSupervisorIOConfig extends SeekableStreamSupervisorIOConfig
+ {
+ TestableSeekableStreamSupervisorIOConfig(
+ String stream,
+ InputFormat inputFormat,
+ Integer replicas,
+ Integer taskCount,
+ Period taskDuration,
+ Period startDelay,
+ Period period,
+ Boolean useEarliestSequenceNumber,
+ Period completionTimeout,
+ Period lateMessageRejectionPeriod,
+ Period earlyMessageRejectionPeriod,
+ AutoScalerConfig autoScalerConfig,
+ LagAggregator lagAggregator,
+ DateTime lateMessageRejectionStartDateTime,
+ IdleConfig idleConfig,
+ Integer stopTaskCount,
+ Map<Integer, Integer> serverPriorityToReplicas,
+ BoundedStreamConfig boundedStreamConfig
+ )
+ {
+ super(
+ stream,
+ inputFormat,
+ replicas,
+ taskCount,
+ taskDuration,
+ startDelay,
+ period,
+ useEarliestSequenceNumber,
+ completionTimeout,
+ lateMessageRejectionPeriod,
+ earlyMessageRejectionPeriod,
+ autoScalerConfig,
+ lagAggregator,
+ lateMessageRejectionStartDateTime,
+ idleConfig,
+ stopTaskCount,
+ serverPriorityToReplicas,
+ boundedStreamConfig
+ );
+ }
+
+ @Override
+ public SupervisorIOConfigBuilder<?, ?> toBuilder()
+ {
+ return new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder().copyFromBase(this);
+ }
+ }
}
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 80120d07fdf..2b54f720e52 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
@@ -26,6 +26,7 @@ import com.google.common.collect.ImmutableMap;
import org.apache.druid.data.input.impl.JsonInputFormat;
import org.apache.druid.error.DruidException;
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.SupervisorStateManager;
@@ -57,6 +58,7 @@ import javax.annotation.Nullable;
import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
@@ -672,28 +674,18 @@ public class SeekableStreamSupervisorSpecTest extends SeekableStreamSupervisorTe
EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();
// Use an ioConfig whose autoScalerConfig matches the test's intent (min=15, max=20) so the
// supervisor's bounds and the scaler's bounds agree.
- EasyMock.expect(spec.getIoConfig()).andReturn(new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- null,
- new Period("PT1H"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- misconfiguredAutoScalerConfig,
- LagAggregator.DEFAULT,
- null,
- null,
- null,
- null,
- null
- )
- {
- }).anyTimes();
+ EasyMock.expect(spec.getIoConfig()).andReturn(new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withAutoScalerConfig(misconfiguredAutoScalerConfig)
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .build()).anyTimes();
EasyMock.expect(spec.getTuningConfig()).andReturn(getTuningConfig()).anyTimes();
EasyMock.expect(spec.getEmitter()).andReturn(emitter).anyTimes();
EasyMock.expect(spec.isSuspended()).andReturn(false).anyTimes();
@@ -749,28 +741,18 @@ public class SeekableStreamSupervisorSpecTest extends SeekableStreamSupervisorTe
// taskCountMin=2 so scaler's floored output (1) is below min and triggers the clamp.
final Map<String, Object> scaleInProps = getScaleInProperties();
scaleInProps.put("taskCountMin", 2);
- final SeekableStreamSupervisorIOConfig customIoConfig = new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- null,
- new Period("PT1H"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- mapper.convertValue(scaleInProps, AutoScalerConfig.class),
- LagAggregator.DEFAULT,
- null,
- null,
- null,
- null,
- null
- )
- {
- };
+ final SeekableStreamSupervisorIOConfig customIoConfig = new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withAutoScalerConfig(mapper.convertValue(scaleInProps, AutoScalerConfig.class))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .build();
EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();
EasyMock.expect(spec.getIoConfig()).andReturn(customIoConfig).anyTimes();
@@ -829,33 +811,23 @@ public class SeekableStreamSupervisorSpecTest extends SeekableStreamSupervisorTe
@Test
public void testSeekableStreamSupervisorSpecWithScaleDisable() throws InterruptedException
{
- SeekableStreamSupervisorIOConfig seekableStreamSupervisorIOConfig = new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- 1,
- new Period("PT1H"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- LagAggregator.DEFAULT,
- null,
- null,
- null,
- null,
- null
- )
- {
- };
+ SeekableStreamSupervisorIOConfig ioConfig = new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .build();
EasyMock.expect(spec.getId()).andReturn(SUPERVISOR).anyTimes();
EasyMock.expect(spec.getSupervisorStateManagerConfig()).andReturn(supervisorConfig).anyTimes();
EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();
- EasyMock.expect(spec.getIoConfig()).andReturn(seekableStreamSupervisorIOConfig).anyTimes();
+ EasyMock.expect(spec.getIoConfig()).andReturn(ioConfig).anyTimes();
EasyMock.expect(spec.getTuningConfig()).andReturn(getTuningConfig()).anyTimes();
EasyMock.expect(spec.getEmitter()).andReturn(emitter).anyTimes();
EasyMock.expect(spec.isSuspended()).andReturn(false).anyTimes();
@@ -888,30 +860,21 @@ public class SeekableStreamSupervisorSpecTest extends SeekableStreamSupervisorTe
@Test
public void testEnablingIdleBeviourPerSupervisorWithOverlordConfigEnabled()
{
- SeekableStreamSupervisorIOConfig seekableStreamSupervisorIOConfig = new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- 1,
- new Period("PT1H"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- LagAggregator.DEFAULT,
- null,
- new IdleConfig(true, null),
- null,
- null,
- null
- )
- {
- };
-
- EasyMock.expect(ingestionSchema.getIOConfig()).andReturn(seekableStreamSupervisorIOConfig).anyTimes();
+ SeekableStreamSupervisorIOConfig ioConfig = new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .withIdleConfig(new IdleConfig(true, null))
+ .build();
+
+ EasyMock.expect(ingestionSchema.getIOConfig()).andReturn(ioConfig).anyTimes();
EasyMock.expect(ingestionSchema.getDataSchema()).andReturn(dataSchema).anyTimes();
EasyMock.replay(ingestionSchema);
spec = new SeekableStreamSupervisorSpec(
@@ -937,21 +900,27 @@ public class SeekableStreamSupervisorSpecTest extends SeekableStreamSupervisorTe
}
@Override
- protected SeekableStreamSupervisorSpec toggleSuspend(boolean suspend)
+ protected SeekableStreamSupervisorSpec toggleSuspend(final boolean suspend)
{
return null;
}
@Override
public SeekableStreamSupervisorSpec createBackfillSpec(
- String backfillId,
- BoundedStreamConfig boundedStreamConfig,
- @Nullable Integer taskCount
+ final String backfillId,
+ final BoundedStreamConfig boundedStreamConfig,
+ @Nullable final Integer taskCount
)
{
return null;
}
+ @Override
+ public Builder<?> toBuilder()
+ {
+ throw new UnsupportedOperationException();
+ }
+
@Override
public String getType()
{
@@ -1454,6 +1423,97 @@ public class SeekableStreamSupervisorSpecTest extends SeekableStreamSupervisorTe
assertMergeResult(null, spec(null, 2, 8, null), 2); // taskCountMin from constructor
}
+ @Test
+ public void testRequireRestart_identicalSpecsDoNotRestart()
+ {
+ final TestSeekableStreamSupervisorSpec seed =
+ 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));
+ }
+
+ @Test
+ public void testRequireRestart_taskCountChangeWithAutoscalerDoesNotRestart()
+ {
+ final TestSeekableStreamSupervisorSpec seed =
+ 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));
+ }
+
+ @Test
+ public void testRequireRestart_taskCountChangeWithoutAutoscalerRestarts()
+ {
+ 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));
+ }
+
+ @Test
+ public void testRequireRestart_contextChangeRestarts()
+ {
+ final TestSeekableStreamSupervisorSpec seed =
+ 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));
+ }
+
+ @Test
+ public void testRequireRestart_suspendedChangeRestarts()
+ {
+ final TestSeekableStreamSupervisorSpec seed =
+ 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));
+ }
+
+ @Test
+ public void testRequireRestart_dataSchemaChangeRestarts()
+ {
+ final TestSeekableStreamSupervisorSpec seed =
+ 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));
+ }
+
+ @Test
+ public void testRequireRestart_differentSpecTypeRestarts()
+ {
+ final TestSeekableStreamSupervisorSpec seed =
+ buildSpecWithIoConfig("id", createIOConfig(2, lagBasedAutoScalerConfig(1, 8, null)));
+ Assert.assertTrue(seed.toBuilder().build().requireRestart(new NoopSupervisorSpec("id", List.of("ds"))));
+ }
+
+ @Test
+ public void testRequireRestart_disablingAutoscalerWithTaskCountChangeRestarts()
+ {
+ // The old spec has autoscaling enabled, so taskCount is neutralized in the comparison. The new
+ // spec disables autoscaling and bumps taskCount; the autoScalerConfig difference must still force
+ // a restart even though the taskCount difference alone would not.
+ final SeekableStreamSupervisorSpec oldSpec =
+ 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));
+ }
+
+ @Test
+ public void testRequireRestart_tuningConfigChangeRestarts()
+ {
+ // No autoscaler, so taskCount is not neutralized; only the tuningConfig differs between the specs.
+ final TestSeekableStreamSupervisorSpec seed = buildSpecWithIoConfig("id", createIOConfig(2, null));
+ final SeekableStreamSupervisorSpec oldSpec = seed.toBuilder().build();
+ final SeekableStreamSupervisorSpec newSpec =
+ seed.toBuilder().tuningConfig(EasyMock.mock(SeekableStreamSupervisorTuningConfig.class)).build();
+ Assert.assertTrue(oldSpec.requireRestart(newSpec));
+ }
+
private void assertMergeResult(
@Nullable TestSeekableStreamSupervisorSpec existingSpec,
TestSeekableStreamSupervisorSpec newSpec,
@@ -1527,53 +1587,24 @@ public class SeekableStreamSupervisorSpecTest extends SeekableStreamSupervisorTe
private SeekableStreamSupervisorIOConfig getIOConfig(boolean scaleOut, int maxTaskCount)
{
- if (scaleOut) {
- return new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- null, // autoscaler uses taskCountStart/taskCountMin for the initial value
- new Period("PT1H"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- mapper.convertValue(getScaleOutProperties(maxTaskCount), AutoScalerConfig.class),
- LagAggregator.DEFAULT,
- null,
- null,
- null,
- null,
- null
- )
- {
- };
- } else {
- return new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- null, // autoscaler uses taskCountStart/taskCountMin for the initial value
- new Period("PT1H"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- mapper.convertValue(getScaleInProperties(), AutoScalerConfig.class),
- LagAggregator.DEFAULT,
- null,
- null,
- null,
- null,
- null
- )
- {
- };
- }
+ final AutoScalerConfig autoScalerConfig = mapper.convertValue(
+ scaleOut ? getScaleOutProperties(maxTaskCount) : getScaleInProperties(),
+ AutoScalerConfig.class
+ );
+ // taskCount is left unset; with autoscaling enabled the ioConfig derives it from
+ // taskCountStart/taskCountMin.
+ return new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withAutoScalerConfig(autoScalerConfig)
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .build();
}
private static DataSchema getDataSchema()
@@ -1630,28 +1661,19 @@ public class SeekableStreamSupervisorSpecTest extends SeekableStreamSupervisorTe
Map<String, Object> endOffsets = ImmutableMap.of("0", 100L);
BoundedStreamConfig boundedConfig = new BoundedStreamConfig(startOffsets, endOffsets);
- SeekableStreamSupervisorIOConfig boundedIoConfig = new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- 1,
- new Period("PT1H"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- LagAggregator.DEFAULT,
- null,
- null,
- null,
- null,
- boundedConfig
- )
- {
- };
+ SeekableStreamSupervisorIOConfig boundedIoConfig = new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .withBoundedStreamConfig(boundedConfig)
+ .build();
EasyMock.expect(spec.getId()).andReturn(SUPERVISOR).anyTimes();
EasyMock.expect(spec.getSupervisorStateManagerConfig()).andReturn(supervisorConfig).anyTimes();
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
index 6c12c8b3885..695a916383e 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java
@@ -686,28 +686,19 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
EasyMock.reset(spec);
EasyMock.expect(spec.isSuspended()).andReturn(false).anyTimes();
EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();
- EasyMock.expect(spec.getIoConfig()).andReturn(new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- 1,
- new Period("PT1H"),
- new Period("PT1S"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- LagAggregator.DEFAULT,
- null,
- new IdleConfig(true, 200L),
- null,
- null,
- null
- )
- {
- }).anyTimes();
+ EasyMock.expect(spec.getIoConfig()).andReturn(new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("PT1S"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .withIdleConfig(new IdleConfig(true, 200L))
+ .build()).anyTimes();
EasyMock.expect(spec.getId()).andReturn(SUPERVISOR_ID).anyTimes();
EasyMock.expect(spec.getTuningConfig()).andReturn(getTuningConfig()).anyTimes();
EasyMock.expect(spec.getEmitter()).andReturn(emitter).anyTimes();
@@ -795,28 +786,19 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
EasyMock.expect(spec.isSuspended()).andReturn(false).anyTimes();
EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();
EasyMock.expect(spec.getContextValue("tags")).andReturn("").anyTimes();
- EasyMock.expect(spec.getIoConfig()).andReturn(new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- 1,
- new Period("PT1H"),
- new Period("PT1S"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- LagAggregator.DEFAULT,
- null,
- new IdleConfig(true, 200L),
- null,
- null,
- null
- )
- {
- }).anyTimes();
+ EasyMock.expect(spec.getIoConfig()).andReturn(new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("PT1S"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .withIdleConfig(new IdleConfig(true, 200L))
+ .build()).anyTimes();
EasyMock.expect(spec.getTuningConfig()).andReturn(getTuningConfig()).anyTimes();
EasyMock.expect(spec.getEmitter()).andReturn(emitter).anyTimes();
EasyMock.expect(spec.getMonitorSchedulerConfig()).andReturn(new DruidMonitorSchedulerConfig()
@@ -1096,26 +1078,19 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
public void testCheckpointForActiveTaskGroup() throws InterruptedException, JsonProcessingException
{
DateTime startTime = DateTimes.nowUtc();
- SeekableStreamSupervisorIOConfig ioConfig = new SeekableStreamSupervisorIOConfig(
- STREAM,
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- 1,
- new Period("PT1H"),
- new Period("PT1S"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- LagAggregator.DEFAULT,
- null,
- new IdleConfig(true, 200L),
- null,
- null,
- null
- ) {};
+ SeekableStreamSupervisorIOConfig ioConfig = new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("PT1S"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .withIdleConfig(new IdleConfig(true, 200L))
+ .build();
EasyMock.reset(spec);
EasyMock.expect(spec.getId()).andReturn(SUPERVISOR_ID).anyTimes();
@@ -1316,28 +1291,20 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
DateTime startTime = DateTimes.nowUtc().minusHours(2);
// Configure supervisor to stop only one task at a time
int stopTaskCount = 1;
- SeekableStreamSupervisorIOConfig ioConfig = new SeekableStreamSupervisorIOConfig(
- STREAM,
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- 3,
- new Period("PT1H"),
- new Period("PT1S"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- LagAggregator.DEFAULT,
- null,
- new IdleConfig(true, 200L),
- stopTaskCount,
- null,
- null
- )
- {
- };
+ SeekableStreamSupervisorIOConfig ioConfig = new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(3)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("PT1S"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .withIdleConfig(new IdleConfig(true, 200L))
+ .withStopTaskCount(stopTaskCount)
+ .build();
EasyMock.reset(spec);
EasyMock.expect(spec.getId()).andReturn(SUPERVISOR_ID).anyTimes();
@@ -1553,28 +1520,19 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
public void testSupervisorStopTaskGroupEarly() throws JsonProcessingException, InterruptedException
{
DateTime startTime = DateTimes.nowUtc();
- SeekableStreamSupervisorIOConfig ioConfig = new SeekableStreamSupervisorIOConfig(
- STREAM,
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- 1,
- new Period("PT1H"),
- new Period("PT1S"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- LagAggregator.DEFAULT,
- null,
- new IdleConfig(true, 200L),
- null,
- null,
- null
- )
- {
- };
+ SeekableStreamSupervisorIOConfig ioConfig = new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("PT1S"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .withIdleConfig(new IdleConfig(true, 200L))
+ .build();
EasyMock.reset(spec);
EasyMock.expect(spec.getId()).andReturn(SUPERVISOR_ID).anyTimes();
@@ -2814,28 +2772,18 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
EasyMock.expect(spec.getId()).andReturn(SUPERVISOR_ID).anyTimes();
EasyMock.expect(spec.getSupervisorStateManagerConfig()).andReturn(supervisorConfig).anyTimes();
EasyMock.expect(spec.getDataSchema()).andReturn(getDataSchema()).anyTimes();
- EasyMock.expect(spec.getIoConfig()).andReturn(new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- 1,
- new Period("PT1H"),
- new Period("PT1S"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- null,
- LagAggregator.DEFAULT,
- null,
- null,
- null,
- null,
- null
- )
- {
- }).anyTimes();
+ EasyMock.expect(spec.getIoConfig()).andReturn(new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("PT1S"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .build()).anyTimes();
EasyMock.expect(spec.getTuningConfig()).andReturn(getTuningConfig()).anyTimes();
EasyMock.expect(spec.getEmitter()).andReturn(emitter).anyTimes();
EasyMock.expect(spec.getMonitorSchedulerConfig()).andReturn(new DruidMonitorSchedulerConfig() {
@@ -2881,28 +2829,18 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
null,
0.4
);
- SeekableStreamSupervisorIOConfig config = new SeekableStreamSupervisorIOConfig(
- "stream",
- null,
- 1,
- null,
- new Period("PT1H"),
- new Period("PT1S"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- autoScalerConfig,
- lagAggregator,
- null,
- null,
- 1, // ensure this is overridden
- null,
- null
- )
- {
- };
+ SeekableStreamSupervisorIOConfig config = new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withReplicas(1)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("PT1S"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withAutoScalerConfig(autoScalerConfig)
+ .withLagAggregator(lagAggregator)
+ .withStopTaskCount(1) // ensure this is overridden
+ .build();
Assert.assertEquals(2, config.getMaxAllowedStops());
config.setTaskCount(30);
@@ -3200,28 +3138,22 @@ public class SeekableStreamSupervisorStateTest extends EasyMockSupport
@Nullable Map<Integer, Integer> serverPriorityToReplicas
)
{
- return new SeekableStreamSupervisorIOConfig(
- "stream",
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- serverPriorityToReplicas == null ? 1 : serverPriorityToReplicas.values().stream().mapToInt(Integer::intValue).sum(),
- taskCount,
- new Period("PT1H"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- autoScalerConfig,
- LagAggregator.DEFAULT,
- null,
- null,
- null,
- serverPriorityToReplicas,
- null
- )
- {
- };
+ return new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream("stream")
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(serverPriorityToReplicas == null
+ ? 1
+ : serverPriorityToReplicas.values().stream().mapToInt(Integer::intValue).sum())
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withAutoScalerConfig(autoScalerConfig)
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .withServerPriorityToReplicas(serverPriorityToReplicas)
+ .build();
}
private static Map<String, Object> getProperties()
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java
index c96a64211b9..968e8a75092 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorTestBase.java
@@ -21,7 +21,6 @@ package org.apache.druid.indexing.seekablestream.supervisor;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableList;
-import com.google.common.collect.ImmutableMap;
import org.apache.druid.data.input.impl.ByteEntity;
import org.apache.druid.data.input.impl.DimensionSchema;
import org.apache.druid.data.input.impl.JsonInputFormat;
@@ -446,6 +445,54 @@ public abstract class SeekableStreamSupervisorTestBase
{
return null;
}
+
+ @Override
+ public Builder toBuilder()
+ {
+ return new Builder().supervisor(supervisor).copyFrom(this);
+ }
+
+ static class Builder extends SeekableStreamSupervisorSpec.Builder<Builder>
+ {
+ private SeekableStreamSupervisor supervisor;
+
+ Builder supervisor(SeekableStreamSupervisor supervisor)
+ {
+ this.supervisor = supervisor;
+ return this;
+ }
+
+ @Override
+ protected Builder self()
+ {
+ return this;
+ }
+
+ @Override
+ public TestSeekableStreamSupervisorSpec build()
+ {
+ final SeekableStreamSupervisorIngestionSpec ingestionSchema =
+ new SeekableStreamSupervisorIngestionSpec(dataSchema, ioConfig, tuningConfig)
+ {
+ };
+ return new TestSeekableStreamSupervisorSpec(
+ ingestionSchema,
+ context,
+ suspended,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ supervisor,
+ id
+ );
+ }
+ }
}
protected static SeekableStreamSupervisorTuningConfig getTuningConfig()
@@ -559,28 +606,19 @@ public abstract class SeekableStreamSupervisorTestBase
AutoScalerConfig autoScalerConfig
)
{
- return new SeekableStreamSupervisorIOConfig(
- STREAM,
- new JsonInputFormat(new JSONPathSpec(true, ImmutableList.of()), ImmutableMap.of(), false, false, false),
- 1,
- taskCount,
- new Period("PT1H"),
- new Period("P1D"),
- new Period("PT30S"),
- false,
- new Period("PT30M"),
- null,
- null,
- autoScalerConfig,
- LagAggregator.DEFAULT,
- null,
- null,
- null,
- null,
- null
- )
- {
- };
+ return new SupervisorIOConfigBuilder.DefaultSupervisorIOConfigBuilder()
+ .withStream(STREAM)
+ .withInputFormat(new JsonInputFormat(new JSONPathSpec(true, List.of()), Map.of(), false, false, false))
+ .withReplicas(1)
+ .withTaskCount(taskCount)
+ .withTaskDuration(new Period("PT1H"))
+ .withStartDelay(new Period("P1D"))
+ .withSupervisorRunPeriod(new Period("PT30S"))
+ .withUseEarliestSequenceNumber(false)
+ .withCompletionTimeout(new Period("PT30M"))
+ .withAutoScalerConfig(autoScalerConfig)
+ .withLagAggregator(LagAggregator.DEFAULT)
+ .build();
}
public static AutoScalerConfig lagBasedAutoScalerConfig(int taskCountMin, int taskCountMax, Integer taskCountStart)
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 67575057ceb..95e20c7858c 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
@@ -139,4 +139,13 @@ public interface SupervisorSpec
{
// No-op by default
}
+
+ /**
+ * Returns true if replacing this (running) spec with {@code proposedSpec} requires a supervisor restart.
+ * Invoked on the running spec; default is conservative (always restart).
+ */
+ default boolean requireRestart(SupervisorSpec proposedSpec)
+ {
+ return true;
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org