Fix serde of UpdateLocationAction used by K8s-based tasks (#18853)

* Fix serde of UpdateLocationAction used by K8s-based tasks
* Add null check and throw exception

Author: kfaraz

indexing-service/src/main/java/org/apache/druid/indexing/common/actions/UpdateLocationAction.java

@@ -23,26 +23,34 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.google.common.base.Optional;
+import org.apache.druid.error.InvalidInput;
 import org.apache.druid.indexer.TaskLocation;
 import org.apache.druid.indexing.common.task.Task;
 import org.apache.druid.indexing.overlord.TaskRunner;
 
+import java.util.Objects;
+
+/**
+ * Notifies the Overlord that the location of a task is now updated.
+ * Used when running K8s-based tasks in encapsulated mode.
+ */
 public class UpdateLocationAction implements TaskAction<Void>
 {
-  private final TaskLocation taskLocation;
+  private final TaskLocation location;
 
   @JsonCreator
   public UpdateLocationAction(
       @JsonProperty("location") TaskLocation location
   )
   {
-    this.taskLocation = location;
+    InvalidInput.conditionalException(location != null, "No task location specified");
+    this.location = location;
   }
 
   @JsonProperty
-  public TaskLocation getTaskLocation()
+  public TaskLocation getLocation()
   {
-    return taskLocation;
+    return location;
   }
 
   @Override
@@ -56,7 +64,7 @@ public Void perform(Task task, TaskActionToolbox toolbox)
   {
     Optional<TaskRunner> taskRunner = toolbox.getTaskRunner();
     if (taskRunner.isPresent()) {
-      taskRunner.get().updateLocation(task, taskLocation);
+      taskRunner.get().updateLocation(task, location);
     }
     return null;
   }
@@ -65,7 +73,26 @@ public Void perform(Task task, TaskActionToolbox toolbox)
   public String toString()
   {
     return "UpdateLocationAction{" +
-           "taskLocation=" + taskLocation +
+           "taskLocation=" + location +
            '}';
   }
+
+  @Override
+  public boolean equals(Object object)
+  {
+    if (this == object) {
+      return true;
+    }
+    if (object == null || getClass() != object.getClass()) {
+      return false;
+    }
+    UpdateLocationAction that = (UpdateLocationAction) object;
+    return Objects.equals(location, that.location);
+  }
+
+  @Override
+  public int hashCode()
+  {
+    return Objects.hashCode(location);
+  }
 }

indexing-service/src/test/java/org/apache/druid/indexing/common/actions/UpdateLocationActionTest.java

@@ -19,11 +19,15 @@
 
 package org.apache.druid.indexing.common.actions;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.google.common.base.Optional;
+import org.apache.druid.error.DruidException;
 import org.apache.druid.indexer.TaskLocation;
 import org.apache.druid.indexing.common.task.NoopTask;
 import org.apache.druid.indexing.common.task.Task;
 import org.apache.druid.indexing.overlord.TaskRunner;
+import org.apache.druid.segment.TestHelper;
+import org.junit.Assert;
 import org.junit.Test;
 
 import java.net.InetAddress;
@@ -68,4 +72,24 @@ public void testWithNoTaskRunner() throws UnknownHostException
     action.perform(task, toolbox);
     verify(runner, never()).updateStatus(any(), any());
   }
+
+  @Test
+  public void testSerde() throws JsonProcessingException
+  {
+    final UpdateLocationAction original = new UpdateLocationAction(
+        TaskLocation.create("host", 1000, -1, false, "pod")
+    );
+    final String json = TestHelper.JSON_MAPPER.writeValueAsString(original);
+    final TaskAction<?> deserialized = TestHelper.JSON_MAPPER.readValue(json, TaskAction.class);
+    Assert.assertEquals(original, deserialized);
+  }
+
+  @Test
+  public void test_actionWithNullLocation_throwsException()
+  {
+    Assert.assertThrows(
+        DruidException.class,
+        () -> new UpdateLocationAction(null)
+    );
+  }
 }