refactor: make supervisor diff+restart algorithm smarter on submission

embedded-tests/src/test/java/org/apache/druid/testing/embedded/docker/IngestionBackwardCompatibilityDockerTest.java

@@ -36,6 +36,7 @@
 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 @@ protected void validateSupervisorUpdateResponse(Map<String, String> startSupervi
     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()
   {

embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/IngestionSmokeTest.java

@@ -59,6 +59,7 @@
 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 void test_runKafkaSupervisor()
     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 void test_streamLogs_ofCancelledTask() throws Exception
   }
 
   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 @@ private KafkaSupervisorSpec createKafkaSupervisor(String topic)
             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 @@ private void waitForSegmentsToBeQueryable(int numSegments)
 
   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()

extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfig.java

@@ -32,6 +32,7 @@
 import javax.annotation.Nullable;
 
 import java.io.File;
+import java.util.Objects;
 
 public class RabbitStreamIndexTaskTuningConfig extends SeekableStreamIndexTaskTuningConfig
 {
@@ -238,6 +239,30 @@ public RabbitStreamIndexTaskTuningConfig withBasePersistDirectory(File dir)
         );
   }
 
+  @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()
   {

extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamIOConfigBuilder.java

@@ -0,0 +1,91 @@
+/*
+ * 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.rabbitstream.supervisor;
+
+import org.apache.druid.indexing.seekablestream.supervisor.SupervisorIOConfigBuilder;
+
+import java.util.Map;
+
+/**
+ * Builder for {@link RabbitStreamSupervisorIOConfig}.
+ */
+public class RabbitStreamIOConfigBuilder
+    extends SupervisorIOConfigBuilder<RabbitStreamIOConfigBuilder, RabbitStreamSupervisorIOConfig>
+{
+  private String uri;
+  private Map<String, Object> consumerProperties;
+  private Long pollTimeout;
+
+  public RabbitStreamIOConfigBuilder withUri(String uri)
+  {
+    this.uri = uri;
+    return this;
+  }
+
+  public RabbitStreamIOConfigBuilder withConsumerProperties(Map<String, Object> consumerProperties)
+  {
+    this.consumerProperties = consumerProperties;
+    return this;
+  }
+
+  public RabbitStreamIOConfigBuilder withPollTimeout(Long pollTimeout)
+  {
+    this.pollTimeout = pollTimeout;
+    return this;
+  }
+
+  /**
+   * Populates this builder (base and Rabbit-specific fields) from an existing config.
+   */
+  public RabbitStreamIOConfigBuilder copyFrom(RabbitStreamSupervisorIOConfig io)
+  {
+    copyFromBase(io);
+    this.uri = io.getUri();
+    this.consumerProperties = io.getConsumerProperties();
+    this.pollTimeout = io.getPollTimeout();
+    return this;
+  }
+
+  @Override
+  public RabbitStreamSupervisorIOConfig build()
+  {
+    return new RabbitStreamSupervisorIOConfig(
+        stream,
+        uri,
+        inputFormat,
+        replicas,
+        taskCount,
+        taskDuration,
+        consumerProperties,
+        autoScalerConfig,
+        pollTimeout,
+        startDelay,
+        period,
+        completionTimeout,
+        useEarliestSequenceNumber,
+        lateMessageRejectionPeriod,
+        earlyMessageRejectionPeriod,
+        lateMessageRejectionStartDateTime,
+        stopTaskCount,
+        serverPriorityToReplicas,
+        boundedStreamConfig
+    );
+  }
+}

extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorIOConfig.java

@@ -33,6 +33,7 @@
 
 import javax.annotation.Nullable;
 import java.util.Map;
+import java.util.Objects;
 
 public class RabbitStreamSupervisorIOConfig extends SeekableStreamSupervisorIOConfig
 {
@@ -145,4 +146,31 @@ public String toString()
         '}';
   }
 
+  @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);
+  }
+
 }

extensions-contrib/rabbit-stream-indexing-service/src/main/java/org/apache/druid/indexing/rabbitstream/supervisor/RabbitStreamSupervisorSpec.java

@@ -216,4 +216,42 @@ public String toString()
         ", 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
+      );
+    }
+  }
 }

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.Period;
 
 import javax.annotation.Nullable;
+import java.util.Objects;
 
 public class RabbitStreamSupervisorTuningConfig extends RabbitStreamIndexTaskTuningConfig
     implements SeekableStreamSupervisorTuningConfig
@@ -185,6 +186,29 @@ public Duration getOffsetFetchPeriod()
     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()
   {

extensions-contrib/rabbit-stream-indexing-service/src/test/java/org/apache/druid/indexing/rabbitstream/RabbitStreamIndexTaskTuningConfigTest.java

@@ -21,6 +21,8 @@
 
 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 void testtoString() throws Exception
 
     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();
+  }
 }