Update the Supervisor endpoint to not restart the Supervisor if the spec was unmodified

indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java

@@ -166,6 +166,18 @@ public boolean createOrUpdateAndStartSupervisor(SupervisorSpec spec)
     }
   }
 
+  public boolean wasSupervisorSpecModified(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");
+      return metadataSupervisorManager.wasSupervisorSpecModified(spec);
+    }
+  }
+
   public boolean stopAndRemoveSupervisor(String id)
   {
     Preconditions.checkState(started, "SupervisorManager not started");

indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResource.java

@@ -56,6 +56,7 @@
 import javax.annotation.Nullable;
 import javax.servlet.http.HttpServletRequest;
 import javax.ws.rs.Consumes;
+import javax.ws.rs.DefaultValue;
 import javax.ws.rs.GET;
 import javax.ws.rs.POST;
 import javax.ws.rs.Path;
@@ -119,7 +120,10 @@ public SupervisorResource(
   @POST
   @Consumes(MediaType.APPLICATION_JSON)
   @Produces(MediaType.APPLICATION_JSON)
-  public Response specPost(final SupervisorSpec spec, @Context final HttpServletRequest req)
+  public Response specPost(
+      final SupervisorSpec spec,
+      @Context final HttpServletRequest req,
+      @QueryParam("restartIfUnmodified") @DefaultValue("true") boolean restartIfUnmodified)
   {
     return asLeaderWithSupervisorManager(
         manager -> {
@@ -151,6 +155,9 @@ public Response specPost(final SupervisorSpec spec, @Context final HttpServletRe
           if (!authResult.allowAccessWithNoRestriction()) {
             throw new ForbiddenException(authResult.getErrorMessage());
           }
+          if (!restartIfUnmodified && !manager.wasSupervisorSpecModified(spec)) {
+            return Response.ok(ImmutableMap.of("id", spec.getId())).build();
+          }
 
           manager.createOrUpdateAndStartSupervisor(spec);
 

indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorResourceTest.java

@@ -161,7 +161,7 @@ public List<String> getDataSources()
 
     replayAll();
 
-    Response response = supervisorResource.specPost(spec, request);
+    Response response = supervisorResource.specPost(spec, request, true);
     verifyAll();
 
     Assert.assertEquals(200, response.getStatus());
@@ -171,7 +171,7 @@ public List<String> getDataSources()
     EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent());
     replayAll();
 
-    response = supervisorResource.specPost(spec, request);
+    response = supervisorResource.specPost(spec, request, true);
     verifyAll();
 
     Assert.assertEquals(503, response.getStatus());
@@ -201,7 +201,7 @@ public List<String> getDataSources()
     EasyMock.expect(authConfig.isEnableInputSourceSecurity()).andReturn(true);
     replayAll();
 
-    Response response = supervisorResource.specPost(spec, request);
+    Response response = supervisorResource.specPost(spec, request, true);
     verifyAll();
 
     Assert.assertEquals(200, response.getStatus());
@@ -211,7 +211,7 @@ public List<String> getDataSources()
     EasyMock.expect(taskMaster.getSupervisorManager()).andReturn(Optional.absent());
     replayAll();
 
-    response = supervisorResource.specPost(spec, request);
+    response = supervisorResource.specPost(spec, request, true);
     verifyAll();
 
     Assert.assertEquals(503, response.getStatus());
@@ -241,7 +241,7 @@ public List<String> getDataSources()
     EasyMock.expect(authConfig.isEnableInputSourceSecurity()).andReturn(true);
     replayAll();
 
-    Assert.assertThrows(ForbiddenException.class, () -> supervisorResource.specPost(spec, request));
+    Assert.assertThrows(ForbiddenException.class, () -> supervisorResource.specPost(spec, request, true));
     verifyAll();
   }
 

server/src/main/java/org/apache/druid/metadata/MetadataSupervisorManager.java

@@ -63,4 +63,11 @@
    * @return number of supervisor removed
    */
   int removeTerminatedSupervisorsOlderThan(long timestamp);
+  /**
+   * Checks whether the submitted spec is different from the spec in the metastore
+   *
+   * @param SupervisorSpec spec being submitted
+   * @return whether the spec was modified
+   */
+  boolean wasSupervisorSpecModified(SupervisorSpec spec);
 }

server/src/main/java/org/apache/druid/metadata/SQLMetadataSupervisorManager.java

@@ -20,6 +20,7 @@
 package org.apache.druid.metadata;
 
 import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonMappingException;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Supplier;
@@ -49,6 +50,7 @@
 import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
@@ -285,4 +287,32 @@ private String getSupervisorsTable()
   {
     return dbTables.get().getSupervisorTable();
   }
+
+  @Override
+  public boolean wasSupervisorSpecModified(SupervisorSpec spec)
+  {
+    byte[] latestSpecAsBytes = dbi.withHandle(
+        handle -> handle.createQuery(
+                            StringUtils.format(
+                                "SELECT id, spec_id, created_date, payload FROM %1$s WHERE spec_id = :spec_id ORDER BY id DESC LIMIT 1",
+                                getSupervisorsTable()
+                            ))
+                        .bind("spec_id", spec.getId())
+                        .map((index, r, ctx) -> r.getBytes("payload"))
+                        .first()
+    );
+    if (latestSpecAsBytes != null) {
+      try {
+        byte[] specAsBytes = jsonMapper.writeValueAsBytes(spec);
+        if (Arrays.equals(specAsBytes, latestSpecAsBytes)) {
+          return false;
+        }
+      }
+      catch (JsonProcessingException ex) {
+        log.warn("Failed to write spec as bytes for spec_id[%s]", spec.getId());
+        return true;
+      }
+    }
+    return true;
+  }
 }