Handle nullable taskTypes for rolling upgrade (#5309)

api/src/main/java/io/druid/indexer/TaskStatusPlus.java

@@ -39,7 +39,7 @@ public class TaskStatusPlus
   @JsonCreator
   public TaskStatusPlus(
       @JsonProperty("id") String id,
-      @JsonProperty("type") String type,
+      @JsonProperty("type") @Nullable String type, // nullable for backward compatibility
       @JsonProperty("createdTime") DateTime createdTime,
       @JsonProperty("queueInsertionTime") DateTime queueInsertionTime,
       @JsonProperty("state") @Nullable TaskState state,
@@ -65,6 +65,7 @@ public class TaskStatusPlus
     return id;
   }
 
+  @Nullable
   @JsonProperty
   public String getType()
   {
@@ -83,12 +84,14 @@ public class TaskStatusPlus
     return queueInsertionTime;
   }
 
+  @Nullable
   @JsonProperty
   public TaskState getState()
   {
     return state;
   }
 
+  @Nullable
   @JsonProperty
   public Long getDuration()
   {

indexing-service/src/main/java/io/druid/indexing/overlord/TaskRunnerWorkItem.java

@@ -27,6 +27,8 @@ import io.druid.indexing.common.TaskStatus;
 import io.druid.java.util.common.DateTimes;
 import org.joda.time.DateTime;
 
+import javax.annotation.Nullable;
+
 /**
  * A holder for a task and different components associated with the task
  */
@@ -85,6 +87,11 @@ public abstract class TaskRunnerWorkItem
   }
 
   public abstract TaskLocation getLocation();
+
+  /**
+   * Returns the type of task. The return value can be null for backward compatibility.
+   */
+  @Nullable
   public abstract String getTaskType();
 
   @Override

indexing-service/src/main/java/io/druid/indexing/overlord/http/OverlordResource.java

@@ -517,7 +517,10 @@ public class OverlordResource
             } else {
               workItems = taskRunner.getRunningTasks()
                                     .stream()
-                                    .filter(workitem -> workitem.getTaskType().equals(taskType))
+                                    .filter(workitem -> {
+                                      final String itemType = workitem.getTaskType();
+                                      return itemType != null && itemType.equals(taskType);
+                                    })
                                     .collect(Collectors.toList());
             }
 

server/src/main/java/io/druid/server/coordinator/helper/DruidCoordinatorSegmentCompactor.java

@@ -72,7 +72,14 @@ public class DruidCoordinatorSegmentCompactor implements DruidCoordinatorHelper
         final int numRunningCompactTasks = indexingServiceClient
             .getRunningTasks()
             .stream()
-            .filter(status -> status.getType().equals(COMPACT_TASK_TYPE))
+            .filter(status -> {
+              final String taskType = status.getType();
+              // taskType can be null if middleManagers are running with an older version. Here, we consevatively regard
+              // the tasks of the unknown taskType as the compactionTask. This is because it's important to not run
+              // compactionTasks more than the configured limit at any time which might impact to the ingestion
+              // performance.
+              return taskType == null || taskType.equals(COMPACT_TASK_TYPE);
+            })
             .collect(Collectors.toList())
             .size();
         final CompactionSegmentIterator iterator = policy.reset(compactionConfigs, dataSources);