Remove index_realtime and index_realtime_appenderator tasks

.github/workflows/revised-its.yml

@@ -50,7 +50,7 @@ jobs:
       matrix:
         #jdk: [8, 11, 17]
         jdk: [8]
-        it: [HighAvailability, MultiStageQuery, Catalog, BatchIndex, MultiStageQueryWithMM, InputSource, InputFormat, Security]
+        it: [HighAvailability, MultiStageQuery, Catalog, BatchIndex, MultiStageQueryWithMM, InputSource, InputFormat, Security, Query]
         #indexer: [indexer, middleManager]
         indexer: [middleManager]
     uses: ./.github/workflows/reusable-revised-its.yml

distribution/bin/tag-missing-milestones.py

@@ -16,10 +16,9 @@
 # limitations under the License.
 
 import os
+import requests
 import subprocess
 import sys
-import requests
-
 
 if len(sys.argv) != 5:
   sys.stderr.write('usage: program <github-username> <previous-release-commit> <new-release-commit> <milestone-number-to-tag>\n')

docs/ingestion/tranquility.md

@@ -22,15 +22,9 @@ title: "Tranquility"
   ~ under the License.
   -->
 
-[Tranquility](https://github.com/druid-io/tranquility/) is a separately distributed package for pushing
-streams to Druid in real-time. 
-
-Tranquility has not been built against a version of Druid later than Druid 0.9.2
-release. It may still work with the latest Druid servers, but not all features and functionality will be available
-due to limitations of older Druid APIs on the Tranquility side.
+[Tranquility](https://github.com/druid-io/tranquility/) was a separately distributed package for pushing
+streams to Druid in real-time. It is not compatible with recent versions of Druid.
 
 For new projects that require streaming ingestion, we recommend using Druid's native support for
 [Apache Kafka](../ingestion/kafka-ingestion.md) or
 [Amazon Kinesis](../ingestion/kinesis-ingestion.md).
-
-For more details, check out the [Tranquility GitHub page](https://github.com/druid-io/tranquility/).

extensions-contrib/materialized-view-maintenance/src/test/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisorTest.java

@@ -64,7 +64,6 @@
 import org.junit.Rule;
 import org.junit.Test;
 
-import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
@@ -145,7 +144,7 @@ public void setUp()
   }
 
   @Test
-  public void testCheckSegments() throws IOException
+  public void testCheckSegments()
   {
     List<DataSegment> baseSegments = createBaseSegments();
     Set<DataSegment> derivativeSegments = Sets.newHashSet(createDerivativeSegments());
@@ -171,7 +170,7 @@ public void testCheckSegments() throws IOException
   }
 
   @Test
-  public void testSubmitTasksDoesNotFailIfTaskAlreadyExists() throws IOException
+  public void testSubmitTasksDoesNotFailIfTaskAlreadyExists()
   {
     Set<DataSegment> baseSegments = Sets.newHashSet(createBaseSegments());
     Set<DataSegment> derivativeSegments = Sets.newHashSet(createDerivativeSegments());
@@ -193,7 +192,7 @@ public void testSubmitTasksDoesNotFailIfTaskAlreadyExists() throws IOException
   }
 
   @Test
-  public void testSubmitTasksFailsIfTaskCannotBeAdded() throws IOException
+  public void testSubmitTasksFailsIfTaskCannotBeAdded()
   {
     Set<DataSegment> baseSegments = Sets.newHashSet(createBaseSegments());
     Set<DataSegment> derivativeSegments = Sets.newHashSet(createDerivativeSegments());
@@ -219,7 +218,7 @@ public void testSubmitTasksFailsIfTaskCannotBeAdded() throws IOException
   }
 
   @Test
-  public void testCheckSegmentsAndSubmitTasks() throws IOException
+  public void testCheckSegmentsAndSubmitTasks()
   {
     Set<DataSegment> baseSegments = Collections.singleton(createBaseSegments().get(0));
     indexerMetadataStorageCoordinator.commitSegments(baseSegments, null);

extensions-contrib/materialized-view-selection/src/test/java/org/apache/druid/query/materializedview/DatasourceOptimizerTest.java

@@ -178,13 +178,8 @@ public void testOptimize() throws InterruptedException
               Lists.newArrayList("dim1", "dim2", "dim3", "dim4"),
               1024 * 1024
           );
-          try {
-            metadataStorageCoordinator.commitSegments(Sets.newHashSet(segment), null);
-            announceSegmentForServer(druidServer, segment, zkPathsConfig, jsonMapper);
-          }
-          catch (IOException e) {
-            return false;
-          }
+          metadataStorageCoordinator.commitSegments(Sets.newHashSet(segment), null);
+          announceSegmentForServer(druidServer, segment, zkPathsConfig, jsonMapper);
           return true;
         }
     );
@@ -203,13 +198,8 @@ public void testOptimize() throws InterruptedException
               Lists.newArrayList("dim1", "dim2", "dim3"),
               1024
           );
-          try {
-            metadataStorageCoordinator.commitSegments(Sets.newHashSet(segment), null);
-            announceSegmentForServer(druidServer, segment, zkPathsConfig, jsonMapper);
-          }
-          catch (IOException e) {
-            return false;
-          }
+          metadataStorageCoordinator.commitSegments(Sets.newHashSet(segment), null);
+          announceSegmentForServer(druidServer, segment, zkPathsConfig, jsonMapper);
           return true;
         }
     );

extensions-core/druid-bloom-filter/src/test/java/org/apache/druid/query/aggregation/bloom/BloomFilterAggregatorTest.java

@@ -49,7 +49,6 @@
 import org.junit.Test;
 
 import javax.annotation.Nullable;
-
 import java.io.IOException;
 import java.nio.ByteBuffer;
 import java.util.Arrays;

extensions-core/druid-pac4j/src/main/java/org/apache/druid/security/pac4j/Pac4jFilter.java

@@ -41,7 +41,6 @@
 import javax.servlet.ServletResponse;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
-
 import java.io.IOException;
 import java.util.Collection;
 

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java

@@ -367,7 +367,7 @@ public void testRunAfterDataInserted() throws Exception
     // Wait for task to exit
     Assert.assertEquals(TaskState.SUCCESS, future.get().getStatusCode());
     verifyTaskMetrics(task, RowMeters.with().bytes(getTotalSizeOfRecords(2, 5)).totalProcessed(3));
-    Assert.assertTrue(task.getRunner().getFireDepartmentMetrics().isProcessingDone());
+    Assert.assertTrue(task.getRunner().getSegmentGenerationMetrics().isProcessingDone());
 
     // Check published metadata and segments in deep storage
     assertEqualsExceptVersion(
@@ -505,7 +505,7 @@ public void testRunAfterDataInsertedWithLegacyParser() throws Exception
     // Wait for task to exit
     Assert.assertEquals(TaskState.SUCCESS, future.get().getStatusCode());
     verifyTaskMetrics(task, RowMeters.with().bytes(getTotalSizeOfRecords(2, 5)).totalProcessed(3));
-    Assert.assertTrue(task.getRunner().getFireDepartmentMetrics().isProcessingDone());
+    Assert.assertTrue(task.getRunner().getSegmentGenerationMetrics().isProcessingDone());
 
     // Check published metadata and segments in deep storage
     assertEqualsExceptVersion(
@@ -554,7 +554,7 @@ public void testRunBeforeDataInserted() throws Exception
     // Wait for task to exit
     Assert.assertEquals(TaskState.SUCCESS, future.get().getStatusCode());
     verifyTaskMetrics(task, RowMeters.with().bytes(getTotalSizeOfRecords(2, 5)).totalProcessed(3));
-    Assert.assertTrue(task.getRunner().getFireDepartmentMetrics().isProcessingDone());
+    Assert.assertTrue(task.getRunner().getSegmentGenerationMetrics().isProcessingDone());
 
     // Check published metadata and segments in deep storage
     assertEqualsExceptVersion(

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java

@@ -40,7 +40,7 @@
 import org.apache.druid.indexer.TaskStatus;
 import org.apache.druid.indexing.common.TaskInfoProvider;
 import org.apache.druid.indexing.common.TestUtils;
-import org.apache.druid.indexing.common.task.RealtimeIndexTask;
+import org.apache.druid.indexing.common.task.NoopTask;
 import org.apache.druid.indexing.common.task.Task;
 import org.apache.druid.indexing.kafka.KafkaConsumerConfigs;
 import org.apache.druid.indexing.kafka.KafkaDataSourceMetadata;
@@ -89,9 +89,7 @@
 import org.apache.druid.segment.incremental.ParseExceptionReport;
 import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
 import org.apache.druid.segment.indexing.DataSchema;
-import org.apache.druid.segment.indexing.RealtimeIOConfig;
 import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
-import org.apache.druid.segment.realtime.FireDepartment;
 import org.apache.druid.server.metrics.NoopServiceEmitter;
 import org.apache.druid.server.security.Action;
 import org.apache.druid.server.security.Resource;
@@ -1194,15 +1192,13 @@ public void testDontKillTasksWithMismatchedType() throws Exception
     addSomeEvents(1);
 
     // non KafkaIndexTask (don't kill)
-    Task id2 = new RealtimeIndexTask(
+    Task id2 = new NoopTask(
         "id2",
         null,
-        new FireDepartment(
-            dataSchema,
-            new RealtimeIOConfig(null, null),
-            null
-        ),
-        null
+        dataSchema.getDataSource(),
+        100,
+        100,
+        ImmutableMap.of()
     );
 
     List<Task> existingTasks = ImmutableList.of(id2);

extensions-core/kinesis-indexing-service/src/test/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisorTest.java

@@ -37,7 +37,7 @@
 import org.apache.druid.indexer.TaskStatus;
 import org.apache.druid.indexing.common.TaskInfoProvider;
 import org.apache.druid.indexing.common.TestUtils;
-import org.apache.druid.indexing.common.task.RealtimeIndexTask;
+import org.apache.druid.indexing.common.task.NoopTask;
 import org.apache.druid.indexing.common.task.Task;
 import org.apache.druid.indexing.kinesis.KinesisDataSourceMetadata;
 import org.apache.druid.indexing.kinesis.KinesisIndexTask;
@@ -84,9 +84,7 @@
 import org.apache.druid.segment.TestHelper;
 import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
 import org.apache.druid.segment.indexing.DataSchema;
-import org.apache.druid.segment.indexing.RealtimeIOConfig;
 import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
-import org.apache.druid.segment.realtime.FireDepartment;
 import org.apache.druid.server.metrics.NoopServiceEmitter;
 import org.apache.druid.server.security.Action;
 import org.apache.druid.server.security.Resource;
@@ -940,15 +938,13 @@ public void testDontKillTasksWithMismatchedType() throws Exception
     EasyMock.expectLastCall().anyTimes();
 
     // non KinesisIndexTask (don't kill)
-    Task id2 = new RealtimeIndexTask(
+    Task id2 = new NoopTask(
         "id2",
         null,
-        new FireDepartment(
-            dataSchema,
-            new RealtimeIOConfig(null, null),
-            null
-        ),
-        null
+        dataSchema.getDataSource(),
+        100,
+        100,
+        ImmutableMap.of()
     );
 
     List<Task> existingTasks = ImmutableList.of(id2);

extensions-core/multi-stage-query/src/main/java/org/apache/druid/msq/counters/SegmentGeneratorMetricsWrapper.java

@@ -20,15 +20,15 @@
 package org.apache.druid.msq.counters;
 
 import org.apache.druid.msq.indexing.processor.SegmentGeneratorFrameProcessor;
-import org.apache.druid.segment.realtime.FireDepartmentMetrics;
+import org.apache.druid.segment.realtime.SegmentGenerationMetrics;
 
 /**
- * Wrapper around {@link FireDepartmentMetrics} which updates the progress counters while updating its metrics. This
+ * Wrapper around {@link SegmentGenerationMetrics} which updates the progress counters while updating its metrics. This
  * is necessary as the {@link org.apache.druid.segment.realtime.appenderator.BatchAppenderator} used by the
  * {@link SegmentGeneratorFrameProcessor} is not part of the MSQ extension, and hence,
  * cannot update the counters used in MSQ reports as it persists and pushes segments to deep storage.
  */
-public class SegmentGeneratorMetricsWrapper extends FireDepartmentMetrics
+public class SegmentGeneratorMetricsWrapper extends SegmentGenerationMetrics
 {
   private final SegmentGenerationProgressCounter segmentGenerationProgressCounter;
 

extensions-core/multi-stage-query/src/test/java/org/apache/druid/msq/test/CalciteMSQTestsHelper.java

@@ -88,7 +88,6 @@
 import org.mockito.Mockito;
 
 import javax.annotation.Nullable;
-
 import java.io.File;
 import java.util.List;
 import java.util.Set;

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

@@ -19,46 +19,34 @@
 
 package org.apache.druid.indexing.common;
 
-import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import org.apache.druid.indexing.common.stats.TaskRealtimeMetricsMonitor;
 import org.apache.druid.indexing.common.task.Task;
 import org.apache.druid.query.DruidMetrics;
 import org.apache.druid.segment.incremental.RowIngestionMeters;
-import org.apache.druid.segment.realtime.FireDepartment;
-import org.apache.druid.segment.realtime.RealtimeMetricsMonitor;
+import org.apache.druid.segment.realtime.SegmentGenerationMetrics;
 
 public class TaskRealtimeMetricsMonitorBuilder
 {
   private TaskRealtimeMetricsMonitorBuilder()
   {
   }
 
-  public static RealtimeMetricsMonitor build(Task task, FireDepartment fireDepartment)
-  {
-    return new RealtimeMetricsMonitor(
-        ImmutableList.of(fireDepartment),
-        ImmutableMap.of(
-            DruidMetrics.TASK_ID, new String[]{task.getId()},
-            DruidMetrics.TASK_TYPE, new String[]{task.getType()}
-        )
-    );
-  }
-
   public static TaskRealtimeMetricsMonitor build(
       Task task,
-      FireDepartment fireDepartment,
+      SegmentGenerationMetrics metrics,
       RowIngestionMeters meters
   )
   {
     return new TaskRealtimeMetricsMonitor(
-        fireDepartment,
+        metrics,
         meters,
         ImmutableMap.of(
+            DruidMetrics.DATASOURCE, new String[]{task.getDataSource()},
             DruidMetrics.TASK_ID, new String[]{task.getId()},
             DruidMetrics.TASK_TYPE, new String[]{task.getType()},
             DruidMetrics.GROUP_ID, new String[]{task.getGroupId()}
-            ),
+        ),
         task.getContextValue(DruidMetrics.TAGS)
     );
   }