Add CliIndexer process type and initial task runner implementation

core/src/main/java/org/apache/druid/guice/Jerseys.java

@@ -24,14 +24,18 @@
 import com.google.inject.multibindings.Multibinder;
 import org.apache.druid.guice.annotations.JSR311Resource;
 import org.apache.druid.guice.annotations.PublicApi;
+import org.apache.druid.java.util.common.logger.Logger;
 
 /**
  */
 @PublicApi
 public class Jerseys
 {
+  private static final Logger LOG = new Logger(Jerseys.class);
+
   public static void addResource(Binder binder, Class<?> resourceClazz)
   {
+    LOG.info("Adding Jersey resource: " + resourceClazz.getName());
     Multibinder.newSetBinder(binder, new TypeLiteral<Class<?>>(){}, JSR311Resource.class)
                .addBinding()
                .toInstance(resourceClazz);

docs/content/querying/lookups.md

@@ -132,6 +132,8 @@ The configuration is propagated to the query serving processes (Broker / Router
 The query serving processes have an internal API for managing lookups on the process and those are used by the Coordinator.
 The Coordinator periodically checks if any of the processes need to load/drop lookups and updates them appropriately.
 
+Please note that only 2 simultaneous lookup configuration propagation requests can be concurrently handled by a single query serving process. This limit is applied to prevent lookup handling from consuming too many server HTTP connections.
+
 # API for configuring lookups
 
 ## Bulk update

extensions-core/druid-basic-security/src/main/java/org/apache/druid/security/basic/CommonCacheNotifier.java

@@ -68,7 +68,8 @@ public class CommonCacheNotifier
       NodeType.HISTORICAL,
       NodeType.PEON,
       NodeType.ROUTER,
-      NodeType.MIDDLE_MANAGER
+      NodeType.MIDDLE_MANAGER,
+      NodeType.INDEXER
   );
 
   private final DruidNodeDiscoveryProvider discoveryProvider;

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/IncrementalPublishingKafkaIndexTaskRunner.java

@@ -37,6 +37,7 @@
 import org.apache.druid.indexing.seekablestream.common.StreamPartition;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.realtime.firehose.ChatHandlerProvider;
 import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.druid.utils.CircularBuffer;
@@ -72,6 +73,7 @@ public IncrementalPublishingKafkaIndexTaskRunner(
       Optional<ChatHandlerProvider> chatHandlerProvider,
       CircularBuffer<Throwable> savedParseExceptions,
       RowIngestionMetersFactory rowIngestionMetersFactory,
+      AppenderatorsManager appenderatorsManager,
       LockGranularity lockGranularityToUse
   )
   {
@@ -82,6 +84,7 @@ public IncrementalPublishingKafkaIndexTaskRunner(
         chatHandlerProvider,
         savedParseExceptions,
         rowIngestionMetersFactory,
+        appenderatorsManager,
         lockGranularityToUse
     );
     this.task = task;

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/KafkaIndexTask.java

@@ -29,6 +29,7 @@
 import org.apache.druid.indexing.seekablestream.SeekableStreamIndexTask;
 import org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner;
 import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.realtime.firehose.ChatHandlerProvider;
 import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -63,7 +64,8 @@ public KafkaIndexTask(
       @JacksonInject ChatHandlerProvider chatHandlerProvider,
       @JacksonInject AuthorizerMapper authorizerMapper,
       @JacksonInject RowIngestionMetersFactory rowIngestionMetersFactory,
-      @JacksonInject ObjectMapper configMapper
+      @JacksonInject ObjectMapper configMapper,
+      @JacksonInject AppenderatorsManager appenderatorsManager
   )
   {
     super(
@@ -76,7 +78,8 @@ public KafkaIndexTask(
         chatHandlerProvider,
         authorizerMapper,
         rowIngestionMetersFactory,
-        getFormattedGroupId(dataSchema.getDataSource(), TYPE)
+        getFormattedGroupId(dataSchema.getDataSource(), TYPE),
+        appenderatorsManager
     );
     this.configMapper = configMapper;
     this.ioConfig = ioConfig;
@@ -136,6 +139,7 @@ protected SeekableStreamIndexTaskRunner<Integer, Long> createTaskRunner()
         chatHandlerProvider,
         savedParseExceptions,
         rowIngestionMetersFactory,
+        appenderatorsManager,
         lockGranularityToUse
     );
   }

extensions-core/kafka-indexing-service/src/main/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisor.java

@@ -258,7 +258,8 @@ protected List<SeekableStreamIndexTask<Integer, Long>> createIndexTasks(
           null,
           null,
           rowIngestionMetersFactory,
-          sortingMapper
+          sortingMapper,
+          null
       ));
     }
     return taskList;

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

@@ -51,9 +51,9 @@
 import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReportData;
 import org.apache.druid.indexing.common.LockGranularity;
 import org.apache.druid.indexing.common.SegmentLoaderFactory;
+import org.apache.druid.indexing.common.SingleFileTaskReportFileWriter;
 import org.apache.druid.indexing.common.TaskLock;
 import org.apache.druid.indexing.common.TaskReport;
-import org.apache.druid.indexing.common.TaskReportFileWriter;
 import org.apache.druid.indexing.common.TaskToolbox;
 import org.apache.druid.indexing.common.TaskToolboxFactory;
 import org.apache.druid.indexing.common.TestUtils;
@@ -68,6 +68,7 @@
 import org.apache.druid.indexing.common.task.IndexTaskTest;
 import org.apache.druid.indexing.common.task.Task;
 import org.apache.druid.indexing.common.task.Tasks;
+import org.apache.druid.indexing.common.task.TestAppenderatorsManager;
 import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisor;
 import org.apache.druid.indexing.kafka.supervisor.KafkaSupervisorIOConfig;
 import org.apache.druid.indexing.kafka.test.TestBroker;
@@ -142,6 +143,7 @@
 import org.apache.druid.segment.loading.LocalDataSegmentPusher;
 import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
 import org.apache.druid.segment.realtime.appenderator.AppenderatorImpl;
+import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifier;
 import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
 import org.apache.druid.segment.transform.ExpressionTransform;
@@ -230,6 +232,7 @@ public static Iterable<Object[]> constructorFeeder()
   private Long maxTotalRows = null;
   private Period intermediateHandoffPeriod = null;
 
+  private AppenderatorsManager appenderatorsManager;
   private TaskToolboxFactory toolboxFactory;
   private IndexerMetadataStorageCoordinator metadataStorageCoordinator;
   private TaskStorage taskStorage;
@@ -372,6 +375,7 @@ public void setupTest() throws IOException
     topic = getTopicName();
     records = generateRecords(topic);
     reportsFile = File.createTempFile("KafkaIndexTaskTestReports-" + System.currentTimeMillis(), "json");
+    appenderatorsManager = new TestAppenderatorsManager();
     makeToolboxFactory();
   }
 
@@ -2534,7 +2538,8 @@ private KafkaIndexTask createTask(
         null,
         null,
         rowIngestionMetersFactory,
-        OBJECT_MAPPER
+        OBJECT_MAPPER,
+        appenderatorsManager
     );
     task.setPollRetryMs(POLL_RETRY_MS);
     return task;
@@ -2701,6 +2706,7 @@ public void close()
     final LocalDataSegmentPusherConfig dataSegmentPusherConfig = new LocalDataSegmentPusherConfig();
     dataSegmentPusherConfig.storageDirectory = getSegmentDirectory();
     final DataSegmentPusher dataSegmentPusher = new LocalDataSegmentPusher(dataSegmentPusherConfig);
+
     toolboxFactory = new TaskToolboxFactory(
         taskConfig,
         taskActionClientFactory,
@@ -2726,7 +2732,7 @@ public void close()
         EasyMock.createNiceMock(DruidNode.class),
         new LookupNodeService("tier"),
         new DataNodeService("tier", 1, ServerType.INDEXER_EXECUTOR, 0),
-        new TaskReportFileWriter(reportsFile)
+        new SingleFileTaskReportFileWriter(reportsFile)
     );
   }
 

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

@@ -82,6 +82,7 @@
 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.segment.realtime.appenderator.DummyForInjectionAppenderatorsManager;
 import org.apache.druid.server.metrics.DruidMonitorSchedulerConfig;
 import org.apache.druid.server.metrics.ExceptionCapturingServiceEmitter;
 import org.apache.druid.server.metrics.NoopServiceEmitter;
@@ -3556,7 +3557,8 @@ private KafkaIndexTask createKafkaIndexTask(
         null,
         null,
         rowIngestionMetersFactory,
-        objectMapper
+        objectMapper,
+        new DummyForInjectionAppenderatorsManager()
     );
   }
 

extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTask.java

@@ -28,6 +28,7 @@
 import org.apache.druid.indexing.seekablestream.SeekableStreamIndexTask;
 import org.apache.druid.indexing.seekablestream.SeekableStreamIndexTaskRunner;
 import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.realtime.firehose.ChatHandlerProvider;
 import org.apache.druid.server.security.AuthorizerMapper;
 
@@ -50,7 +51,8 @@ public KinesisIndexTask(
       @JacksonInject ChatHandlerProvider chatHandlerProvider,
       @JacksonInject AuthorizerMapper authorizerMapper,
       @JacksonInject RowIngestionMetersFactory rowIngestionMetersFactory,
-      @JacksonInject AWSCredentialsConfig awsCredentialsConfig
+      @JacksonInject AWSCredentialsConfig awsCredentialsConfig,
+      @JacksonInject AppenderatorsManager appenderatorsManager
   )
   {
     super(
@@ -63,7 +65,8 @@ public KinesisIndexTask(
         chatHandlerProvider,
         authorizerMapper,
         rowIngestionMetersFactory,
-        getFormattedGroupId(dataSchema.getDataSource(), TYPE)
+        getFormattedGroupId(dataSchema.getDataSource(), TYPE),
+        appenderatorsManager
     );
     this.awsCredentialsConfig = awsCredentialsConfig;
   }
@@ -79,6 +82,7 @@ protected SeekableStreamIndexTaskRunner<String, String> createTaskRunner()
         chatHandlerProvider,
         savedParseExceptions,
         rowIngestionMetersFactory,
+        appenderatorsManager,
         lockGranularityToUse
     );
   }

extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskRunner.java

@@ -37,6 +37,7 @@
 import org.apache.druid.indexing.seekablestream.common.StreamPartition;
 import org.apache.druid.java.util.common.ISE;
 import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.realtime.firehose.ChatHandlerProvider;
 import org.apache.druid.server.security.AuthorizerMapper;
 import org.apache.druid.utils.CircularBuffer;
@@ -66,6 +67,7 @@ public class KinesisIndexTaskRunner extends SeekableStreamIndexTaskRunner<String
       Optional<ChatHandlerProvider> chatHandlerProvider,
       CircularBuffer<Throwable> savedParseExceptions,
       RowIngestionMetersFactory rowIngestionMetersFactory,
+      AppenderatorsManager appenderatorsManager,
       LockGranularity lockGranularityToUse
   )
   {
@@ -76,6 +78,7 @@ public class KinesisIndexTaskRunner extends SeekableStreamIndexTaskRunner<String
         chatHandlerProvider,
         savedParseExceptions,
         rowIngestionMetersFactory,
+        appenderatorsManager,
         lockGranularityToUse
     );
     this.task = task;

extensions-core/kinesis-indexing-service/src/main/java/org/apache/druid/indexing/kinesis/supervisor/KinesisSupervisor.java

@@ -173,7 +173,8 @@ protected List<SeekableStreamIndexTask<String, String>> createIndexTasks(
           null,
           null,
           rowIngestionMetersFactory,
-          awsCredentialsConfig
+          awsCredentialsConfig,
+          null
       ));
     }
     return taskList;

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

@@ -57,9 +57,9 @@
 import org.apache.druid.indexing.common.IngestionStatsAndErrorsTaskReportData;
 import org.apache.druid.indexing.common.LockGranularity;
 import org.apache.druid.indexing.common.SegmentLoaderFactory;
+import org.apache.druid.indexing.common.SingleFileTaskReportFileWriter;
 import org.apache.druid.indexing.common.TaskLock;
 import org.apache.druid.indexing.common.TaskReport;
-import org.apache.druid.indexing.common.TaskReportFileWriter;
 import org.apache.druid.indexing.common.TaskToolbox;
 import org.apache.druid.indexing.common.TaskToolboxFactory;
 import org.apache.druid.indexing.common.TestUtils;
@@ -75,6 +75,7 @@
 import org.apache.druid.indexing.common.task.Task;
 import org.apache.druid.indexing.common.task.TaskResource;
 import org.apache.druid.indexing.common.task.Tasks;
+import org.apache.druid.indexing.common.task.TestAppenderatorsManager;
 import org.apache.druid.indexing.kinesis.supervisor.KinesisSupervisor;
 import org.apache.druid.indexing.overlord.DataSourceMetadata;
 import org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator;
@@ -141,6 +142,7 @@
 import org.apache.druid.segment.loading.LocalDataSegmentPusher;
 import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
 import org.apache.druid.segment.realtime.appenderator.AppenderatorImpl;
+import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
 import org.apache.druid.segment.realtime.firehose.ChatHandlerProvider;
 import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifier;
 import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
@@ -231,6 +233,7 @@ public static Iterable<Object[]> constructorFeeder()
   private final Period intermediateHandoffPeriod = null;
   private int maxRecordsPerPoll;
 
+  private AppenderatorsManager appenderatorsManager;
   private TaskToolboxFactory toolboxFactory;
   private IndexerMetadataStorageCoordinator metadataStorageCoordinator;
   private TaskStorage taskStorage;
@@ -316,6 +319,8 @@ public void setupTest() throws IOException, InterruptedException
 
     recordSupplier = mock(KinesisRecordSupplier.class);
 
+    appenderatorsManager = new TestAppenderatorsManager();
+
     // sleep required because of kinesalite
     Thread.sleep(500);
     makeToolboxFactory();
@@ -2720,7 +2725,8 @@ private KinesisIndexTask createTask(
         null,
         null,
         rowIngestionMetersFactory,
-        null
+        null,
+        appenderatorsManager
     );
   }
 
@@ -2880,6 +2886,7 @@ public void close()
     final LocalDataSegmentPusherConfig dataSegmentPusherConfig = new LocalDataSegmentPusherConfig();
     dataSegmentPusherConfig.storageDirectory = getSegmentDirectory();
     final DataSegmentPusher dataSegmentPusher = new LocalDataSegmentPusher(dataSegmentPusherConfig);
+
     toolboxFactory = new TaskToolboxFactory(
         taskConfig,
         taskActionClientFactory,
@@ -2905,7 +2912,7 @@ public void close()
         EasyMock.createNiceMock(DruidNode.class),
         new LookupNodeService("tier"),
         new DataNodeService("tier", 1, ServerType.INDEXER_EXECUTOR, 0),
-        new TaskReportFileWriter(reportsFile)
+        new SingleFileTaskReportFileWriter(reportsFile)
     );
   }
 
@@ -3088,7 +3095,8 @@ private TestableKinesisIndexTask(
         @JacksonInject ChatHandlerProvider chatHandlerProvider,
         @JacksonInject AuthorizerMapper authorizerMapper,
         @JacksonInject RowIngestionMetersFactory rowIngestionMetersFactory,
-        @JacksonInject AWSCredentialsConfig awsCredentialsConfig
+        @JacksonInject AWSCredentialsConfig awsCredentialsConfig,
+        @JacksonInject AppenderatorsManager appenderatorsManager
     )
     {
       super(
@@ -3101,7 +3109,8 @@ private TestableKinesisIndexTask(
           chatHandlerProvider,
           authorizerMapper,
           rowIngestionMetersFactory,
-          awsCredentialsConfig
+          awsCredentialsConfig,
+          appenderatorsManager
       );
     }
 

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

@@ -80,6 +80,7 @@
 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.segment.realtime.appenderator.DummyForInjectionAppenderatorsManager;
 import org.apache.druid.server.metrics.DruidMonitorSchedulerConfig;
 import org.apache.druid.server.metrics.ExceptionCapturingServiceEmitter;
 import org.apache.druid.server.metrics.NoopServiceEmitter;
@@ -4259,7 +4260,8 @@ private KinesisIndexTask createKinesisIndexTask(
         null,
         null,
         rowIngestionMetersFactory,
-        null
+        null,
+        new DummyForInjectionAppenderatorsManager()
     );
   }