Add support for concurrent batch Append and Replace

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

@@ -192,8 +192,8 @@ public void testCheckSegments() throws IOException
             1024
         )
     );
-    indexerMetadataStorageCoordinator.announceHistoricalSegments(baseSegments);
-    indexerMetadataStorageCoordinator.announceHistoricalSegments(derivativeSegments);
+    indexerMetadataStorageCoordinator.commitSegments(baseSegments);
+    indexerMetadataStorageCoordinator.commitSegments(derivativeSegments);
     EasyMock.expect(taskMaster.getTaskQueue()).andReturn(Optional.of(taskQueue)).anyTimes();
     EasyMock.expect(taskMaster.getTaskRunner()).andReturn(Optional.absent()).anyTimes();
     EasyMock.expect(taskStorage.getActiveTasks()).andReturn(ImmutableList.of()).anyTimes();
@@ -252,7 +252,7 @@ public void testCheckSegmentsAndSubmitTasks() throws IOException
             1024
         )
     );
-    indexerMetadataStorageCoordinator.announceHistoricalSegments(baseSegments);
+    indexerMetadataStorageCoordinator.commitSegments(baseSegments);
     EasyMock.expect(taskMaster.getTaskQueue()).andReturn(Optional.of(taskQueue)).anyTimes();
     EasyMock.expect(taskMaster.getTaskRunner()).andReturn(Optional.absent()).anyTimes();
     EasyMock.expect(taskStorage.getActiveTasks()).andReturn(ImmutableList.of()).anyTimes();

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

@@ -160,7 +160,7 @@ public void testOptimize() throws InterruptedException
               1024 * 1024
           );
           try {
-            metadataStorageCoordinator.announceHistoricalSegments(Sets.newHashSet(segment));
+            metadataStorageCoordinator.commitSegments(Sets.newHashSet(segment));
             announceSegmentForServer(druidServer, segment, zkPathsConfig, jsonMapper);
           }
           catch (IOException e) {
@@ -185,7 +185,7 @@ public void testOptimize() throws InterruptedException
               1024
           );
           try {
-            metadataStorageCoordinator.announceHistoricalSegments(Sets.newHashSet(segment));
+            metadataStorageCoordinator.commitSegments(Sets.newHashSet(segment));
             announceSegmentForServer(druidServer, segment, zkPathsConfig, jsonMapper);
           }
           catch (IOException e) {

extensions-contrib/sqlserver-metadata-storage/src/test/java/org/apache/druid/metadata/storage/sqlserver/SQLServerConnectorTest.java

@@ -37,19 +37,7 @@ public void testIsTransientException()
     SQLServerConnector connector = new SQLServerConnector(
         Suppliers.ofInstance(new MetadataStorageConnectorConfig()),
         Suppliers.ofInstance(
-            new MetadataStorageTablesConfig(
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null
-            )
+            MetadataStorageTablesConfig.fromBase(null)
         )
     );
 
@@ -70,7 +58,7 @@ public void testLimitClause()
     SQLServerConnector connector = new SQLServerConnector(
         Suppliers.ofInstance(new MetadataStorageConnectorConfig()),
         Suppliers.ofInstance(
-            new MetadataStorageTablesConfig(null, null, null, null, null, null, null, null, null, null, null)
+            MetadataStorageTablesConfig.fromBase(null)
         )
     );
     Assert.assertEquals("FETCH NEXT 100 ROWS ONLY", connector.limitClause(100));

extensions-core/mysql-metadata-storage/src/test/java/org/apache/druid/metadata/storage/mysql/MySQLConnectorTest.java

@@ -45,7 +45,7 @@ public String getDriverClassName()
   private static final Supplier<MetadataStorageConnectorConfig> CONNECTOR_CONFIG_SUPPLIER =
       MetadataStorageConnectorConfig::new;
   private static final Supplier<MetadataStorageTablesConfig> TABLES_CONFIG_SUPPLIER =
-      () -> new MetadataStorageTablesConfig(null, null, null, null, null, null, null, null, null, null, null);
+      () -> MetadataStorageTablesConfig.fromBase(null);
 
 
   @Test

extensions-core/postgresql-metadata-storage/src/test/java/org/apache/druid/metadata/storage/postgresql/PostgreSQLConnectorTest.java

@@ -35,21 +35,7 @@ public void testIsTransientException()
   {
     PostgreSQLConnector connector = new PostgreSQLConnector(
         Suppliers.ofInstance(new MetadataStorageConnectorConfig()),
-        Suppliers.ofInstance(
-            new MetadataStorageTablesConfig(
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null,
-                null
-            )
-        ),
+        Suppliers.ofInstance(MetadataStorageTablesConfig.fromBase(null)),
         new PostgreSQLConnectorConfig(),
         new PostgreSQLTablesConfig()
     );
@@ -68,9 +54,7 @@ public void testLimitClause()
   {
     PostgreSQLConnector connector = new PostgreSQLConnector(
         Suppliers.ofInstance(new MetadataStorageConnectorConfig()),
-        Suppliers.ofInstance(
-            new MetadataStorageTablesConfig(null, null, null, null, null, null, null, null, null, null, null)
-        ),
+        Suppliers.ofInstance(MetadataStorageTablesConfig.fromBase(null)),
         new PostgreSQLConnectorConfig(),
         new PostgreSQLTablesConfig()
     );

indexing-hadoop/src/main/java/org/apache/druid/indexer/updater/MetadataStorageUpdaterJobSpec.java

@@ -90,6 +90,7 @@ public MetadataStorageTablesConfig getMetadataStorageTablesConfig()
         null,
         null,
         null,
+        null,
         segmentTable,
         null,
         null,

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

@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.druid.indexing.common.actions;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.type.TypeReference;
+import org.apache.druid.indexing.common.task.Task;
+import org.apache.druid.indexing.overlord.DataSourceMetadata;
+import org.apache.druid.indexing.overlord.SegmentPublishResult;
+
+import javax.annotation.Nullable;
+
+/**
+ * A stream ingestion task didn't ingest any rows and created no segments (e.g., all records were unparseable),
+ * but still needs to update metadata with the progress that the task made.
+ */
+public class CommitMetadataAction implements TaskAction<SegmentPublishResult>
+{
+  @Nullable
+  private final DataSourceMetadata startMetadata;
+  @Nullable
+  private final DataSourceMetadata endMetadata;
+  @Nullable
+  private final String dataSource;
+
+  public static CommitMetadataAction create(
+      String dataSource,
+      DataSourceMetadata startMetadata,
+      DataSourceMetadata endMetadata
+  )
+  {
+    return new CommitMetadataAction(startMetadata, endMetadata, dataSource);
+  }
+
+  @JsonCreator
+  private CommitMetadataAction(
+      @JsonProperty("startMetadata") @Nullable DataSourceMetadata startMetadata,
+      @JsonProperty("endMetadata") @Nullable DataSourceMetadata endMetadata,
+      @JsonProperty("dataSource") @Nullable String dataSource
+  )
+  {
+    this.startMetadata = startMetadata;
+    this.endMetadata = endMetadata;
+    this.dataSource = dataSource;
+  }
+
+  @JsonProperty
+  @Nullable
+  public DataSourceMetadata getStartMetadata()
+  {
+    return startMetadata;
+  }
+
+  @JsonProperty
+  @Nullable
+  public DataSourceMetadata getEndMetadata()
+  {
+    return endMetadata;
+  }
+
+  @JsonProperty
+  @Nullable
+  public String getDataSource()
+  {
+    return dataSource;
+  }
+
+  @Override
+  public TypeReference<SegmentPublishResult> getReturnTypeReference()
+  {
+    return new TypeReference<SegmentPublishResult>()
+    {
+    };
+  }
+
+  /**
+   * Performs some sanity checks and publishes the given segments.
+   */
+  @Override
+  public SegmentPublishResult perform(Task task, TaskActionToolbox toolbox)
+  {
+    try {
+      return toolbox.getIndexerMetadataStorageCoordinator().commitMetadataOnly(
+          dataSource,
+          startMetadata,
+          endMetadata
+      );
+    }
+    catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+
+  @Override
+  public boolean isAudited()
+  {
+    return true;
+  }
+
+  @Override
+  public String toString()
+  {
+    return "CommitMetadataAction{" +
+           "startMetadata=" + startMetadata +
+           ", endMetadata=" + endMetadata +
+           ", dataSource='" + dataSource +
+           '}';
+  }
+}

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

@@ -30,9 +30,6 @@
 import java.util.Set;
 
 /**
- * Insert segments into metadata storage. The segment versions must all be less than or equal to a lock held by
- * your task for the segment intervals.
- * <p/>
  * Word of warning: Very large "segments" sets can cause oversized audit log entries, which is bad because it means
  * that the task cannot actually complete. Callers should avoid this by avoiding inserting too many segments in the
  * same action.
@@ -65,7 +62,7 @@ public TypeReference<Set<DataSegment>> getReturnTypeReference()
 
   /**
    * Behaves similarly to
-   * {@link org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator#announceHistoricalSegments},
+   * {@link org.apache.druid.indexing.overlord.IndexerMetadataStorageCoordinator#commitSegments},
    * with startMetadata and endMetadata both null.
    */
   @Override

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

@@ -69,7 +69,7 @@ public Void perform(Task task, TaskActionToolbox toolbox)
     try {
       toolbox.getTaskLockbox().doInCriticalSection(
           task,
-          segments.stream().map(DataSegment::getInterval).collect(Collectors.toList()),
+          segments.stream().map(DataSegment::getInterval).collect(Collectors.toSet()),
           CriticalAction.builder()
                         .onValidLocks(
                             () -> {

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

@@ -71,7 +71,7 @@ public Void perform(Task task, TaskActionToolbox toolbox)
     try {
       toolbox.getTaskLockbox().doInCriticalSection(
           task,
-          segments.stream().map(DataSegment::getInterval).collect(Collectors.toList()),
+          segments.stream().map(DataSegment::getInterval).collect(Collectors.toSet()),
           CriticalAction.builder()
                         .onValidLocks(
                             () -> {