You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2021/09/02 04:03:58 UTC

[GitHub] [druid] loquisgon opened a new pull request #11536: Make persists concurrent with adding rows in batch ingestion

loquisgon opened a new pull request #11536:
URL: https://github.com/apache/druid/pull/11536


   Recent testing have shown a potential performance degradation in the bounded memory work for segment creation of batch ingestion.  The reason is most probably the fact that intermediate persists are blocking, serial, with ingestion. We will add back concurrent persists and test performance to validate that previous performance is back.
   
   Changes:
   Intermediate persists are made concurrent with adding rows in the batch appenderator. 
   
   
   This PR has:
   - [ X] been self-reviewed.
      - [x] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
   - [ ] added documentation for new or modified features or behaviors.
   - [ ] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
   - [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
   - [X ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
   - [X ] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
   - [ ] added integration tests.
   - [ ] been tested in a test Druid cluster.
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] lgtm-com[bot] commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-914872827


   This pull request **introduces 1 alert** when merging c85a201fb428ee6b7d728fbefce3864e16d9e322 into dcee99df7871946f1f697817e84f30ce79ea65e3 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-8147cfb1365d27f8a1b37baa18b0d2afd4916b7e)
   
   **new alerts:**
   
   * 1 for Uncontrolled data used in path expression


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] clintropolis merged pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
clintropolis merged pull request #11536:
URL: https://github.com/apache/druid/pull/11536


   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] lgtm-com[bot] commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-914908800


   This pull request **introduces 1 alert** when merging 86973a1975335d7dc1fb2ffb3a87d1337ebba4b6 into dcee99df7871946f1f697817e84f30ce79ea65e3 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-c7cf358492214e81acc30a93dc027dcc5b52831c)
   
   **new alerts:**
   
   * 1 for Uncontrolled data used in path expression


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] lgtm-com[bot] commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-914954118


   This pull request **introduces 1 alert** when merging bd83203d680798e2827053d0cb2986604b87cee9 into dcee99df7871946f1f697817e84f30ce79ea65e3 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-16d6de8634d7b6a085821a62411c00d7c1808b70)
   
   **new alerts:**
   
   * 1 for Uncontrolled data used in path expression


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r703932161



##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java
##########
@@ -117,7 +131,19 @@ public TaskConfig(
       this.shuffleDataLocations = shuffleDataLocations;
     }
     this.ignoreTimestampSpecForDruidInputSource = ignoreTimestampSpecForDruidInputSource;
-    this.useLegacyBatchProcessing = useLegacyBatchProcessing;
+
+    this.batchMemoryMappedIndex = batchMemoryMappedIndex;
+    // Conflict resolution. Assume that if batchMemoryMappedIndex is set (since false is the default) that
+    // the user changed it intentionally to use legacy, in this case oveeride batchProcessingMode and also
+    // set it to legacy else just use batchProcessingMode and don't pay attention to batchMemoryMappedIndexMode:
+    if (batchMemoryMappedIndex) {
+      this.batchProcessingMode = BatchProcesingMode.LEGACY;
+    } else if (EnumUtils.isValidEnum(BatchProcesingMode.class, batchProcesingMode)) {
+      this.batchProcessingMode = BatchProcesingMode.valueOf(batchProcesingMode);
+    } else {
+      // batchProcessingMode input string is invalid, just use the default...log message somewhere???
+      this.batchProcessingMode = BatchProcesingMode.CLOSED_SEGMENTS; // Default
+    }

Review comment:
       yeah




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] clintropolis commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
clintropolis commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r703983991



##########
File path: indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
##########
@@ -314,7 +314,8 @@ public File getTaskReportsFile()
         );
 
         final TaskToolbox box = new TaskToolbox(
-            new TaskConfig(null, null, null, null, null, false, null, null, null, false, false),
+            new TaskConfig(null, null, null, null, null, false, null, null, null, false, false,
+                           TaskConfig.BatchProcessingMode.CLOSED_SEGMENTS.name()),

Review comment:
       nit: should this use default instead of this explicit value?

##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/LegacyAndClosedSegmentsBatchAppenderatorDriverTest.java
##########
@@ -0,0 +1,205 @@
+/*
+ * 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.segment.realtime.appenderator;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.apache.druid.common.config.NullHandling;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.indexing.overlord.SegmentPublishResult;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.segment.loading.DataSegmentKiller;
+import org.apache.druid.segment.realtime.appenderator.BaseAppenderatorDriver.SegmentsForSequence;
+import org.apache.druid.segment.realtime.appenderator.SegmentWithState.SegmentState;
+import org.apache.druid.segment.realtime.appenderator.StreamAppenderatorDriverTest.TestSegmentAllocator;
+import org.apache.druid.timeline.partition.NumberedShardSpec;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockSupport;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class LegacyAndClosedSegmentsBatchAppenderatorDriverTest extends EasyMockSupport

Review comment:
       same super nit replacing Legacy with Open

##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/AppenderatorImpl.java
##########
@@ -0,0 +1,1596 @@
+/*
+ * 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.segment.realtime.appenderator;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Stopwatch;
+import com.google.common.base.Supplier;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.commons.lang.mutable.MutableLong;
+import org.apache.druid.client.cache.Cache;
+import org.apache.druid.data.input.Committer;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.java.util.common.io.Closer;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QuerySegmentWalker;
+import org.apache.druid.query.SegmentDescriptor;
+import org.apache.druid.segment.IndexIO;
+import org.apache.druid.segment.IndexMerger;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.ReferenceCountingSegment;
+import org.apache.druid.segment.Segment;
+import org.apache.druid.segment.incremental.IncrementalIndexAddResult;
+import org.apache.druid.segment.incremental.IndexSizeExceededException;
+import org.apache.druid.segment.incremental.ParseExceptionHandler;
+import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.loading.DataSegmentPusher;
+import org.apache.druid.segment.realtime.FireDepartmentMetrics;
+import org.apache.druid.segment.realtime.FireHydrant;
+import org.apache.druid.segment.realtime.plumber.Sink;
+import org.apache.druid.server.coordination.DataSegmentAnnouncer;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.VersionedIntervalTimeline;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+/**
+ * This class is to support LEGACY and CLOSED_SEGMENTS appenderators. It is copied as-is
+ * from 0.21 and it is meant to keep for backward compatibility. For now though this class

Review comment:
       nit: this isn't actually a copy of 0.21... its of some intermediary state between 0.21 and 0.22. I think `StreamAppenderator` is technically a copy of `ApenderatorImpl` in 0.21.

##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/LegacyAndClosedSegmentsBatchAppenderatorTest.java
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.segment.realtime.appenderator;
+
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.partition.LinearShardSpec;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class LegacyAndClosedSegmentsBatchAppenderatorTest extends InitializedNullHandlingTest

Review comment:
       same nit legacy/open

##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/LegacyAndClosedSegmentsAppenderatorTester.java
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.segment.realtime.appenderator;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.JSONParseSpec;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.core.NoopEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+import org.apache.druid.segment.IndexIO;
+import org.apache.druid.segment.IndexMerger;
+import org.apache.druid.segment.IndexMergerV9;
+import org.apache.druid.segment.IndexSpec;
+import org.apache.druid.segment.column.ColumnConfig;
+import org.apache.druid.segment.incremental.ParseExceptionHandler;
+import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.incremental.SimpleRowIngestionMeters;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.indexing.TuningConfig;
+import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
+import org.apache.druid.segment.loading.DataSegmentPusher;
+import org.apache.druid.segment.realtime.FireDepartmentMetrics;
+import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.partition.LinearShardSpec;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+public class LegacyAndClosedSegmentsAppenderatorTester implements AutoCloseable

Review comment:
       super-nit: `OpenAndClosed...` instead of `LegacyAndClosed`




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] jihoonson commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
jihoonson commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r683867733



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -177,9 +193,53 @@ public Object startJob()
   {
     tuningConfig.getBasePersistDirectory().mkdirs();
     lockBasePersistDirectory();
+    initializeExecutors();
     return null;
   }
 
+  private void throwPersistErrorIfExists()
+  {
+    if (persistError != null) {
+      throw new RE(persistError, "Error while persisting");
+    }
+  }
+
+  private void initializeExecutors()
+  {
+    log.info("There will be up to[%d] pending persists", maxPendingPersists);

Review comment:
       How could this be useful except for debugging? If this is only useful for debugging, it should be not info.

##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
##########
@@ -913,6 +939,10 @@ static InputRow createInputRow(String ts, String dim, Object met)
     );
   }
 
+  private void waitForPersists() throws InterruptedException
+  {
+    Thread.sleep(500);

Review comment:
       Sleeps are bad. These will make the unit testing slower. Also, I bet all the sleeps you added will make these tests quite flaky. 

##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
##########
@@ -806,7 +831,8 @@ public void testTotalRowCount() throws Exception
       appenderator.add(IDENTIFIERS.get(2), createInputRow("2001", "bob", 1), null);
       Assert.assertEquals(4, appenderator.getTotalRowCount());
 
-      appenderator.persistAll(null).get();
+      appenderator.persistAll(null);
+      waitForPersists();

Review comment:
       Why not `persistAll(null).get()`?

##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -110,20 +113,28 @@
   private final ObjectMapper objectMapper;
   private final IndexIO indexIO;
   private final IndexMerger indexMerger;
-  private final Map<SegmentIdWithShardSpec, Sink> sinks = new HashMap<>();
   private final long maxBytesTuningConfig;
   private final boolean skipBytesInMemoryOverheadCheck;
 
+  private volatile ListeningExecutorService persistExecutor = null;
+  private volatile ListeningExecutorService pushExecutor = null;
+  private final int maxPendingPersists;
+  private static final int DEFAULT_PENDING_PERSISTS = 2;
+  private static final int PERSIST_WARN_DELAY = 1000;
+  private volatile Throwable persistError;
+
+
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, Sink> sinks = new ConcurrentHashMap<>();
   /**
    * The following sinks metadata map and associated class are the way to retain metadata now that sinks
    * are being completely removed from memory after each incremental persist.
    */
-  private final Map<SegmentIdWithShardSpec, SinkMetadata> sinksMetadata = new HashMap<>();
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, SinkMetadata> sinksMetadata = new ConcurrentHashMap<>();

Review comment:
       Please document details of the concurrent access pattern.

##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -110,20 +113,28 @@
   private final ObjectMapper objectMapper;
   private final IndexIO indexIO;
   private final IndexMerger indexMerger;
-  private final Map<SegmentIdWithShardSpec, Sink> sinks = new HashMap<>();
   private final long maxBytesTuningConfig;
   private final boolean skipBytesInMemoryOverheadCheck;
 
+  private volatile ListeningExecutorService persistExecutor = null;
+  private volatile ListeningExecutorService pushExecutor = null;
+  private final int maxPendingPersists;
+  private static final int DEFAULT_PENDING_PERSISTS = 2;
+  private static final int PERSIST_WARN_DELAY = 1000;
+  private volatile Throwable persistError;
+
+
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, Sink> sinks = new ConcurrentHashMap<>();
   /**
    * The following sinks metadata map and associated class are the way to retain metadata now that sinks
    * are being completely removed from memory after each incremental persist.
    */
-  private final Map<SegmentIdWithShardSpec, SinkMetadata> sinksMetadata = new HashMap<>();
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, SinkMetadata> sinksMetadata = new ConcurrentHashMap<>();
 
   // This variable updated in add(), persist(), and drop()
-  private int rowsCurrentlyInMemory = 0;
+  private final AtomicInteger rowsCurrentlyInMemory = new AtomicInteger();

Review comment:
       What threads can access this and `bytesCurrentlyInMemory` concurrently?  If there are any, please document details of the concurrent access pattern.

##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -158,6 +169,11 @@
 
     maxBytesTuningConfig = tuningConfig.getMaxBytesInMemoryOrDefault();
     skipBytesInMemoryOverheadCheck = tuningConfig.isSkipBytesInMemoryOverheadCheck();
+    if (tuningConfig.getMaxPendingPersists() < 1) {
+      maxPendingPersists = DEFAULT_PENDING_PERSISTS;

Review comment:
       What is the rationale for the default of 2? The previous default was 0 which is infinite. I don't think we ever need to change this in production. The doc for `maxPendingPersists` was not updated in https://github.com/apache/druid/pull/11294, so whatever we change here, we should fix the doc too.

##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -110,20 +113,28 @@
   private final ObjectMapper objectMapper;
   private final IndexIO indexIO;
   private final IndexMerger indexMerger;
-  private final Map<SegmentIdWithShardSpec, Sink> sinks = new HashMap<>();
   private final long maxBytesTuningConfig;
   private final boolean skipBytesInMemoryOverheadCheck;
 
+  private volatile ListeningExecutorService persistExecutor = null;
+  private volatile ListeningExecutorService pushExecutor = null;
+  private final int maxPendingPersists;
+  private static final int DEFAULT_PENDING_PERSISTS = 2;
+  private static final int PERSIST_WARN_DELAY = 1000;
+  private volatile Throwable persistError;
+
+
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, Sink> sinks = new ConcurrentHashMap<>();

Review comment:
       Can multiple threads access this map at the same time? I don't see any unless I'm missing something. If there are any, please document details of the concurrent access pattern. It helps people a lot including reviewers, other developers, and your future-self to understand and remember how things work. Also please check out https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md. We have reasons for including it in the PR template. I would highly suggest reading it and marking the concurrency self-review item in your PR checklist.

##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
##########
@@ -382,12 +387,14 @@ public void testTaskDoesNotFailAsExceededMemoryWithSkipBytesInMemoryOverheadChec
 
       appenderator.startJob();
       appenderator.add(IDENTIFIERS.get(0), createInputRow("2000", "foo", 1), null);
+      waitForPersists();

Review comment:
       I think you could keep the future of persist triggered in `add()` in a variable. Then you can add a method used only for testing that returns the persist future. Then you can finally wait for the future to be done instead of sleeping.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686296526



##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
##########
@@ -382,12 +387,14 @@ public void testTaskDoesNotFailAsExceededMemoryWithSkipBytesInMemoryOverheadChec
 
       appenderator.startJob();
       appenderator.add(IDENTIFIERS.get(0), createInputRow("2000", "foo", 1), null);
+      waitForPersists();

Review comment:
       I was trying to avoid more changes to the class (exposing methods etc) but I can pursue your suggestion




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686265933



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -177,9 +193,53 @@ public Object startJob()
   {
     tuningConfig.getBasePersistDirectory().mkdirs();
     lockBasePersistDirectory();
+    initializeExecutors();
     return null;
   }
 
+  private void throwPersistErrorIfExists()
+  {
+    if (persistError != null) {
+      throw new RE(persistError, "Error while persisting");
+    }
+  }
+
+  private void initializeExecutors()
+  {
+    log.info("There will be up to[%d] pending persists", maxPendingPersists);

Review comment:
       I'll change it to debug




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r702196466



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManager.java
##########
@@ -227,8 +226,81 @@ public Appenderator createOfflineAppenderatorForTask(
           indexIO,
           wrapIndexMerger(indexMerger),
           rowIngestionMeters,
-          parseExceptionHandler,
-          useLegacyBatchProcessing
+          parseExceptionHandler
+      );
+      datasourceBundle.addAppenderator(taskId, appenderator);
+      return appenderator;
+    }
+  }
+
+  @Override
+  public Appenderator createLegacyOfflineAppenderatorForTask(
+      String taskId,
+      DataSchema schema,
+      AppenderatorConfig config,
+      FireDepartmentMetrics metrics,
+      DataSegmentPusher dataSegmentPusher,
+      ObjectMapper objectMapper,
+      IndexIO indexIO,
+      IndexMerger indexMerger,
+      RowIngestionMeters rowIngestionMeters,
+      ParseExceptionHandler parseExceptionHandler
+  )
+  {
+    synchronized (this) {
+      DatasourceBundle datasourceBundle = datasourceBundles.computeIfAbsent(
+          schema.getDataSource(),
+          DatasourceBundle::new
+      );
+
+      Appenderator appenderator = Appenderators.createLegacyOffline(
+          taskId,
+          schema,
+          rewriteAppenderatorConfigMemoryLimits(config),
+          metrics,
+          dataSegmentPusher,
+          objectMapper,
+          indexIO,
+          wrapIndexMerger(indexMerger),
+          rowIngestionMeters,
+          parseExceptionHandler
+      );
+      datasourceBundle.addAppenderator(taskId, appenderator);
+      return appenderator;
+    }
+  }
+
+  @Override
+  public Appenderator createClosedSegmentsOfflineAppenderatorForTask(
+      String taskId,
+      DataSchema schema,
+      AppenderatorConfig config,
+      FireDepartmentMetrics metrics,
+      DataSegmentPusher dataSegmentPusher,
+      ObjectMapper objectMapper,
+      IndexIO indexIO,
+      IndexMerger indexMerger,
+      RowIngestionMeters rowIngestionMeters,
+      ParseExceptionHandler parseExceptionHandler
+  )
+  {
+    synchronized (this) {
+      DatasourceBundle datasourceBundle = datasourceBundles.computeIfAbsent(
+          schema.getDataSource(),
+          DatasourceBundle::new
+      );
+
+      Appenderator appenderator = Appenderators.createLegacyOffline(

Review comment:
       This is wrong, it should be createClosedSegmentsOffline, I will fix it




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686443280



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -110,20 +113,28 @@
   private final ObjectMapper objectMapper;
   private final IndexIO indexIO;
   private final IndexMerger indexMerger;
-  private final Map<SegmentIdWithShardSpec, Sink> sinks = new HashMap<>();
   private final long maxBytesTuningConfig;
   private final boolean skipBytesInMemoryOverheadCheck;
 
+  private volatile ListeningExecutorService persistExecutor = null;
+  private volatile ListeningExecutorService pushExecutor = null;
+  private final int maxPendingPersists;
+  private static final int DEFAULT_PENDING_PERSISTS = 2;
+  private static final int PERSIST_WARN_DELAY = 1000;
+  private volatile Throwable persistError;
+
+
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, Sink> sinks = new ConcurrentHashMap<>();
   /**
    * The following sinks metadata map and associated class are the way to retain metadata now that sinks
    * are being completely removed from memory after each incremental persist.
    */
-  private final Map<SegmentIdWithShardSpec, SinkMetadata> sinksMetadata = new HashMap<>();
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, SinkMetadata> sinksMetadata = new ConcurrentHashMap<>();

Review comment:
       Added comments about the usage for this in the javadoc (next commit)




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon closed pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon closed pull request #11536:
URL: https://github.com/apache/druid/pull/11536


   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-911869537


   Changes I added in the latest commit: 
   
   - Make sure close() waits for persists & pushes before closing
   - Merged master branch to pick up fixes to a previous issue (serialization/deserialization of JSON interval when it contains UTC timestamps)
   - Renamed `BatchAppenderator#clearSinkMetadata` to `BatchAppenderator#clearSinkMemoryCountersAndDiskStoredData` to make intent clearer


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] jon-wei commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
jon-wei commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r703732837



##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java
##########
@@ -43,6 +44,15 @@
       "org.apache.hadoop:hadoop-client:2.8.5"
   );
 
+  public enum BatchProcesingMode

Review comment:
       BatchProcesingMode -> BatchProcessingMode




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-892950491


   I replaced the semaphore using the style of concurrency used in the appenderator. Now persists and push are ran in the background.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686295009



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -110,20 +113,28 @@
   private final ObjectMapper objectMapper;
   private final IndexIO indexIO;
   private final IndexMerger indexMerger;
-  private final Map<SegmentIdWithShardSpec, Sink> sinks = new HashMap<>();
   private final long maxBytesTuningConfig;
   private final boolean skipBytesInMemoryOverheadCheck;
 
+  private volatile ListeningExecutorService persistExecutor = null;
+  private volatile ListeningExecutorService pushExecutor = null;
+  private final int maxPendingPersists;
+  private static final int DEFAULT_PENDING_PERSISTS = 2;
+  private static final int PERSIST_WARN_DELAY = 1000;
+  private volatile Throwable persistError;
+
+
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, Sink> sinks = new ConcurrentHashMap<>();
   /**
    * The following sinks metadata map and associated class are the way to retain metadata now that sinks
    * are being completely removed from memory after each incremental persist.
    */
-  private final Map<SegmentIdWithShardSpec, SinkMetadata> sinksMetadata = new HashMap<>();
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, SinkMetadata> sinksMetadata = new ConcurrentHashMap<>();
 
   // This variable updated in add(), persist(), and drop()
-  private int rowsCurrentlyInMemory = 0;
+  private final AtomicInteger rowsCurrentlyInMemory = new AtomicInteger();

Review comment:
       Removed unnecessary atomics in next committ.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686295728



##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
##########
@@ -913,6 +939,10 @@ static InputRow createInputRow(String ts, String dim, Object met)
     );
   }
 
+  private void waitForPersists() throws InterruptedException
+  {
+    Thread.sleep(500);

Review comment:
       oh I see your suggestion below...




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] lgtm-com[bot] commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-912877043


   This pull request **introduces 1 alert** when merging 37e0856521ee2d8b9e6e3c2431c2a528f0999d3a into 59d257816b85dbeeca336b8e25d341d67bbc5697 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-60c697d30171cf79639bc5a503ed51f6cabd74a4)
   
   **new alerts:**
   
   * 1 for Uncontrolled data used in path expression


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r703998083



##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/LegacyAndClosedSegmentsAppenderatorTester.java
##########
@@ -0,0 +1,293 @@
+/*
+ * 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.segment.realtime.appenderator;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.druid.data.input.impl.DimensionsSpec;
+import org.apache.druid.data.input.impl.JSONParseSpec;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.jackson.DefaultObjectMapper;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.java.util.emitter.core.NoopEmitter;
+import org.apache.druid.java.util.emitter.service.ServiceEmitter;
+import org.apache.druid.query.aggregation.AggregatorFactory;
+import org.apache.druid.query.aggregation.CountAggregatorFactory;
+import org.apache.druid.query.aggregation.LongSumAggregatorFactory;
+import org.apache.druid.segment.IndexIO;
+import org.apache.druid.segment.IndexMerger;
+import org.apache.druid.segment.IndexMergerV9;
+import org.apache.druid.segment.IndexSpec;
+import org.apache.druid.segment.column.ColumnConfig;
+import org.apache.druid.segment.incremental.ParseExceptionHandler;
+import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.incremental.SimpleRowIngestionMeters;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.indexing.TuningConfig;
+import org.apache.druid.segment.indexing.granularity.UniformGranularitySpec;
+import org.apache.druid.segment.loading.DataSegmentPusher;
+import org.apache.druid.segment.realtime.FireDepartmentMetrics;
+import org.apache.druid.segment.writeout.OffHeapMemorySegmentWriteOutMediumFactory;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.partition.LinearShardSpec;
+
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
+
+public class LegacyAndClosedSegmentsAppenderatorTester implements AutoCloseable

Review comment:
       yeah




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] clintropolis commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
clintropolis commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r704050643



##########
File path: docs/configuration/index.md
##########
@@ -1343,7 +1343,7 @@ Additional peon configs include:
 |`druid.peon.mode`|Choices are "local" and "remote". Setting this to local means you intend to run the peon as a standalone process (Not recommended).|remote|
 |`druid.indexer.task.baseDir`|Base temporary working directory.|`System.getProperty("java.io.tmpdir")`|
 |`druid.indexer.task.baseTaskDir`|Base temporary working directory for tasks.|`${druid.indexer.task.baseDir}/persistent/task`|
-|`druid.indexer.task.useLegacyBatchProcessing`|If false, native batch ingestion will use a new, recommended, code path with memory optimized code for the segment creation phase. If true it will use the previous code path for the create segments phase of batch ingestion. This does not apply to streaming ingestion, just to batch. This setting should only be used when a bug is suspected or found in the new optimized batch ingestion code. If a bug is suspected or found, you can set this flag to `true` to fall back to previous, working but more memory intensive, code path.|`false`|
+|`druid.indexer.task.batchProcessingMode`| Batch ingestion tasks have three operating modes that control how intermediary segments are constructed and tracked: `OPEN_SEGMENTS`, `CLOSED_SEGMENTS`, and `CLOSED_SEGMENT_SINKS`. `OPEN_SEGMENTS` will use code based on the original batch ingestion path and performs a `mmap` on intermediary segments to build a timeline so that these segments can be queryable by realtime queries. This is not needed at all for batch, so the default mode, `CLOSED_SEGMENTS`, eliminates `mmap` of intermediary segments, but still tracks the entire set of segments in heap. The `CLOSED_SEGMENTS_SINKS` mode is the most aggressive and should have the smallest memory footprint, and works by eliminating in memory tracking and mmap of intermediary segments produced during segment creation. This mode isn't as well tested as other modes so is currently considered experimental. `OPEN_SEGMENTS` mode can be selected if any problems occur with the 2 newer modes. |`CLOSED_SEGM
 ENTS`|

Review comment:
       ```suggestion
   |`druid.indexer.task.batchProcessingMode`| Batch ingestion tasks have three operating modes that control how intermediary segments are constructed and tracked: `OPEN_SEGMENTS`, `CLOSED_SEGMENTS`, and `CLOSED_SEGMENT_SINKS`. `OPEN_SEGMENTS` will use code based on the original batch ingestion path and performs a `mmap` on intermediary segments to build a timeline so that these segments can be queryable by realtime queries. This is not needed at all for batch, so the default mode, `CLOSED_SEGMENTS`, eliminates `mmap` of intermediary segments, but still tracks the entire set of segments in heap. The `CLOSED_SEGMENTS_SINKS` mode is the most aggressive and should have the smallest memory footprint, and works by eliminating in memory tracking and `mmap` of intermediary segments produced during segment creation. This mode isn't as well tested as other modes so is currently considered experimental. `OPEN_SEGMENTS` mode can be selected if any problems occur with the 2 newer modes. |`CLOSED_
 SEGMENTS`|
   ```
   should fix spellcheck CI failure i think




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r703998106



##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/LegacyAndClosedSegmentsBatchAppenderatorDriverTest.java
##########
@@ -0,0 +1,205 @@
+/*
+ * 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.segment.realtime.appenderator;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.ImmutableSet;
+import org.apache.druid.common.config.NullHandling;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.indexing.overlord.SegmentPublishResult;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.java.util.common.granularity.Granularities;
+import org.apache.druid.segment.loading.DataSegmentKiller;
+import org.apache.druid.segment.realtime.appenderator.BaseAppenderatorDriver.SegmentsForSequence;
+import org.apache.druid.segment.realtime.appenderator.SegmentWithState.SegmentState;
+import org.apache.druid.segment.realtime.appenderator.StreamAppenderatorDriverTest.TestSegmentAllocator;
+import org.apache.druid.timeline.partition.NumberedShardSpec;
+import org.easymock.EasyMock;
+import org.easymock.EasyMockSupport;
+import org.junit.After;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+public class LegacyAndClosedSegmentsBatchAppenderatorDriverTest extends EasyMockSupport

Review comment:
       check

##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/LegacyAndClosedSegmentsBatchAppenderatorTest.java
##########
@@ -0,0 +1,229 @@
+/*
+ * 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.segment.realtime.appenderator;
+
+import com.google.common.base.Function;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.common.collect.Lists;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.MapBasedInputRow;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.Intervals;
+import org.apache.druid.testing.InitializedNullHandlingTest;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.partition.LinearShardSpec;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class LegacyAndClosedSegmentsBatchAppenderatorTest extends InitializedNullHandlingTest

Review comment:
       check




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r703997776



##########
File path: indexing-service/src/test/java/org/apache/druid/indexing/common/task/IngestionTestBase.java
##########
@@ -314,7 +314,8 @@ public File getTaskReportsFile()
         );
 
         final TaskToolbox box = new TaskToolbox(
-            new TaskConfig(null, null, null, null, null, false, null, null, null, false, false),
+            new TaskConfig(null, null, null, null, null, false, null, null, null, false, false,
+                           TaskConfig.BatchProcessingMode.CLOSED_SEGMENTS.name()),

Review comment:
       done in next 




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] jihoonson commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
jihoonson commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-893936593


   BTW, I assumed that the core logic is not changed and reviewed only the concurrency part. Let me know if this is not true.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686179188



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -158,6 +169,11 @@
 
     maxBytesTuningConfig = tuningConfig.getMaxBytesInMemoryOrDefault();
     skipBytesInMemoryOverheadCheck = tuningConfig.isSkipBytesInMemoryOverheadCheck();
+    if (tuningConfig.getMaxPendingPersists() < 1) {
+      maxPendingPersists = DEFAULT_PENDING_PERSISTS;

Review comment:
       It just seems reasonable to me... I can go back to the previous default, I have no issues with that.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] lgtm-com[bot] commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-915494483


   This pull request **introduces 1 alert** when merging d97615f1799a255bcf767bd5f46dc0cf6b590e70 into dcee99df7871946f1f697817e84f30ce79ea65e3 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-670de71711a08465121dd8e3ca6adde8279049c6)
   
   **new alerts:**
   
   * 1 for Uncontrolled data used in path expression


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] lgtm-com[bot] commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-915363979


   This pull request **introduces 1 alert** when merging 7f004f1880b7778f67b3bb3d9253e803aa6b57cd into dcee99df7871946f1f697817e84f30ce79ea65e3 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-89588882c4a6e4834018cb6cbc3608c30ca6ff94)
   
   **new alerts:**
   
   * 1 for Uncontrolled data used in path expression


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-892950491


   I replaced the semaphore using the style of concurrency used in the appenderator. Now persists and push are ran in the background.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-896182248


   You are correct, @jihoonson, core logic is not changed.


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r702198963



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/UnifiedIndexerAppenderatorsManager.java
##########
@@ -227,8 +226,81 @@ public Appenderator createOfflineAppenderatorForTask(
           indexIO,
           wrapIndexMerger(indexMerger),
           rowIngestionMeters,
-          parseExceptionHandler,
-          useLegacyBatchProcessing
+          parseExceptionHandler
+      );
+      datasourceBundle.addAppenderator(taskId, appenderator);
+      return appenderator;
+    }
+  }
+
+  @Override
+  public Appenderator createLegacyOfflineAppenderatorForTask(
+      String taskId,
+      DataSchema schema,
+      AppenderatorConfig config,
+      FireDepartmentMetrics metrics,
+      DataSegmentPusher dataSegmentPusher,
+      ObjectMapper objectMapper,
+      IndexIO indexIO,
+      IndexMerger indexMerger,
+      RowIngestionMeters rowIngestionMeters,
+      ParseExceptionHandler parseExceptionHandler
+  )
+  {
+    synchronized (this) {
+      DatasourceBundle datasourceBundle = datasourceBundles.computeIfAbsent(
+          schema.getDataSource(),
+          DatasourceBundle::new
+      );
+
+      Appenderator appenderator = Appenderators.createLegacyOffline(
+          taskId,
+          schema,
+          rewriteAppenderatorConfigMemoryLimits(config),
+          metrics,
+          dataSegmentPusher,
+          objectMapper,
+          indexIO,
+          wrapIndexMerger(indexMerger),
+          rowIngestionMeters,
+          parseExceptionHandler
+      );
+      datasourceBundle.addAppenderator(taskId, appenderator);
+      return appenderator;
+    }
+  }
+
+  @Override
+  public Appenderator createClosedSegmentsOfflineAppenderatorForTask(
+      String taskId,
+      DataSchema schema,
+      AppenderatorConfig config,
+      FireDepartmentMetrics metrics,
+      DataSegmentPusher dataSegmentPusher,
+      ObjectMapper objectMapper,
+      IndexIO indexIO,
+      IndexMerger indexMerger,
+      RowIngestionMeters rowIngestionMeters,
+      ParseExceptionHandler parseExceptionHandler
+  )
+  {
+    synchronized (this) {
+      DatasourceBundle datasourceBundle = datasourceBundles.computeIfAbsent(
+          schema.getDataSource(),
+          DatasourceBundle::new
+      );
+
+      Appenderator appenderator = Appenderators.createLegacyOffline(

Review comment:
       Done




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686444349



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -110,20 +113,28 @@
   private final ObjectMapper objectMapper;
   private final IndexIO indexIO;
   private final IndexMerger indexMerger;
-  private final Map<SegmentIdWithShardSpec, Sink> sinks = new HashMap<>();
   private final long maxBytesTuningConfig;
   private final boolean skipBytesInMemoryOverheadCheck;
 
+  private volatile ListeningExecutorService persistExecutor = null;
+  private volatile ListeningExecutorService pushExecutor = null;
+  private final int maxPendingPersists;
+  private static final int DEFAULT_PENDING_PERSISTS = 2;
+  private static final int PERSIST_WARN_DELAY = 1000;
+  private volatile Throwable persistError;
+
+
+  private final ConcurrentHashMap<SegmentIdWithShardSpec, Sink> sinks = new ConcurrentHashMap<>();

Review comment:
       No concurrency control required for this map, removed in next commit.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] lgtm-com[bot] commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-914509462


   This pull request **introduces 1 alert** when merging 1b1fd7b7269336fbc995714473e7d4a690b2132b into 60efbb51d0282c1af9c132e7bd37a43e69867622 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-b677470c4b1fe6dc1a74e44bb0b732760c3e53da)
   
   **new alerts:**
   
   * 1 for Uncontrolled data used in path expression


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] imply-jhan commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
imply-jhan commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r703832810



##########
File path: docs/configuration/index.md
##########
@@ -1343,7 +1343,8 @@ Additional peon configs include:
 |`druid.peon.mode`|Choices are "local" and "remote". Setting this to local means you intend to run the peon as a standalone process (Not recommended).|remote|
 |`druid.indexer.task.baseDir`|Base temporary working directory.|`System.getProperty("java.io.tmpdir")`|
 |`druid.indexer.task.baseTaskDir`|Base temporary working directory for tasks.|`${druid.indexer.task.baseDir}/persistent/task`|
-|`druid.indexer.task.useLegacyBatchProcessing`|If false, native batch ingestion will use a new, recommended, code path with memory optimized code for the segment creation phase. If true it will use the previous code path for the create segments phase of batch ingestion. This does not apply to streaming ingestion, just to batch. This setting should only be used when a bug is suspected or found in the new optimized batch ingestion code. If a bug is suspected or found, you can set this flag to `true` to fall back to previous, working but more memory intensive, code path.|`false`|
+|`druid.indexer.task.batchMemoryMappedIndex`|DEPRECATED: Use `druid.indexer.task.batchProcessingMode` instead. If false, native batch ingestion will use a new, recommended, code path with memory optimized code for the segment creation phase. If true it will use the previous code path for the create segments phase of batch ingestion. This does not apply to streaming ingestion, just to batch. This setting should only be used when a bug is suspected or found in the new optimized batch ingestion code. If a bug is suspected or found, you can set this flag to `true` to fall back to previous, working but more memory intensive, code path.|`false`|

Review comment:
       Maybe call out that set batchMemoryMappedIndex to true will set batchProcessingMode to LEGACY and overwrite the batchProcessingMode value.

##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/config/TaskConfig.java
##########
@@ -117,7 +131,19 @@ public TaskConfig(
       this.shuffleDataLocations = shuffleDataLocations;
     }
     this.ignoreTimestampSpecForDruidInputSource = ignoreTimestampSpecForDruidInputSource;
-    this.useLegacyBatchProcessing = useLegacyBatchProcessing;
+
+    this.batchMemoryMappedIndex = batchMemoryMappedIndex;
+    // Conflict resolution. Assume that if batchMemoryMappedIndex is set (since false is the default) that
+    // the user changed it intentionally to use legacy, in this case oveeride batchProcessingMode and also
+    // set it to legacy else just use batchProcessingMode and don't pay attention to batchMemoryMappedIndexMode:
+    if (batchMemoryMappedIndex) {
+      this.batchProcessingMode = BatchProcesingMode.LEGACY;
+    } else if (EnumUtils.isValidEnum(BatchProcesingMode.class, batchProcesingMode)) {
+      this.batchProcessingMode = BatchProcesingMode.valueOf(batchProcesingMode);
+    } else {
+      // batchProcessingMode input string is invalid, just use the default...log message somewhere???
+      this.batchProcessingMode = BatchProcesingMode.CLOSED_SEGMENTS; // Default
+    }

Review comment:
       Can we log the batchProcessingMode value after this if block?




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r703997841



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/AppenderatorImpl.java
##########
@@ -0,0 +1,1596 @@
+/*
+ * 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.segment.realtime.appenderator;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Function;
+import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
+import com.google.common.base.Stopwatch;
+import com.google.common.base.Supplier;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.google.common.primitives.Ints;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
+import org.apache.commons.lang.mutable.MutableLong;
+import org.apache.druid.client.cache.Cache;
+import org.apache.druid.data.input.Committer;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.java.util.common.DateTimes;
+import org.apache.druid.java.util.common.FileUtils;
+import org.apache.druid.java.util.common.IAE;
+import org.apache.druid.java.util.common.ISE;
+import org.apache.druid.java.util.common.Pair;
+import org.apache.druid.java.util.common.RE;
+import org.apache.druid.java.util.common.RetryUtils;
+import org.apache.druid.java.util.common.StringUtils;
+import org.apache.druid.java.util.common.concurrent.Execs;
+import org.apache.druid.java.util.common.io.Closer;
+import org.apache.druid.java.util.emitter.EmittingLogger;
+import org.apache.druid.query.Query;
+import org.apache.druid.query.QueryRunner;
+import org.apache.druid.query.QuerySegmentWalker;
+import org.apache.druid.query.SegmentDescriptor;
+import org.apache.druid.segment.IndexIO;
+import org.apache.druid.segment.IndexMerger;
+import org.apache.druid.segment.QueryableIndex;
+import org.apache.druid.segment.QueryableIndexSegment;
+import org.apache.druid.segment.ReferenceCountingSegment;
+import org.apache.druid.segment.Segment;
+import org.apache.druid.segment.incremental.IncrementalIndexAddResult;
+import org.apache.druid.segment.incremental.IndexSizeExceededException;
+import org.apache.druid.segment.incremental.ParseExceptionHandler;
+import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.loading.DataSegmentPusher;
+import org.apache.druid.segment.realtime.FireDepartmentMetrics;
+import org.apache.druid.segment.realtime.FireHydrant;
+import org.apache.druid.segment.realtime.plumber.Sink;
+import org.apache.druid.server.coordination.DataSegmentAnnouncer;
+import org.apache.druid.timeline.DataSegment;
+import org.apache.druid.timeline.SegmentId;
+import org.apache.druid.timeline.VersionedIntervalTimeline;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.io.Closeable;
+import java.io.File;
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.file.StandardOpenOption;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
+import java.util.stream.Collectors;
+
+/**
+ * This class is to support LEGACY and CLOSED_SEGMENTS appenderators. It is copied as-is
+ * from 0.21 and it is meant to keep for backward compatibility. For now though this class

Review comment:
       Updated in next




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686179188



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -158,6 +169,11 @@
 
     maxBytesTuningConfig = tuningConfig.getMaxBytesInMemoryOrDefault();
     skipBytesInMemoryOverheadCheck = tuningConfig.isSkipBytesInMemoryOverheadCheck();
+    if (tuningConfig.getMaxPendingPersists() < 1) {
+      maxPendingPersists = DEFAULT_PENDING_PERSISTS;

Review comment:
       I see it is already documented in external docs that its default is zero. Good catch thanks!




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686179188



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -158,6 +169,11 @@
 
     maxBytesTuningConfig = tuningConfig.getMaxBytesInMemoryOrDefault();
     skipBytesInMemoryOverheadCheck = tuningConfig.isSkipBytesInMemoryOverheadCheck();
+    if (tuningConfig.getMaxPendingPersists() < 1) {
+      maxPendingPersists = DEFAULT_PENDING_PERSISTS;

Review comment:
       It just seems reasonable to me... I can go back to the previous default, I have no issues with that. Besides it is already documented in external docs that its default is zero. Good catch thanks!




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon edited a comment on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon edited a comment on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-911869537


   Changes I added in the latest commit (507c2f6): 
   
   - Make sure close() waits for persists & pushes before closing
   - Merged master branch to pick up fixes to a previous issue (serialization/deserialization of JSON interval when it contains UTC timestamps)
   - Renamed `BatchAppenderator#clearSinkMetadata` to `BatchAppenderator#clearSinkMemoryCountersAndDiskStoredData` to make intent clearer


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686295259



##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
##########
@@ -806,7 +831,8 @@ public void testTotalRowCount() throws Exception
       appenderator.add(IDENTIFIERS.get(2), createInputRow("2001", "bob", 1), null);
       Assert.assertEquals(4, appenderator.getTotalRowCount());
 
-      appenderator.persistAll(null).get();
+      appenderator.persistAll(null);
+      waitForPersists();

Review comment:
       Typo --- fixed in next commit.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686185379



##########
File path: server/src/test/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderatorTest.java
##########
@@ -913,6 +939,10 @@ static InputRow createInputRow(String ts, String dim, Object met)
     );
   }
 
+  private void waitForPersists() throws InterruptedException
+  {
+    Thread.sleep(500);

Review comment:
       The use case here is somehow wait for a concurrent task to check on its effect, it seems like a legitimate use for sleep. Let me know what you are thinking that could be better.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686265933



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -177,9 +193,53 @@ public Object startJob()
   {
     tuningConfig.getBasePersistDirectory().mkdirs();
     lockBasePersistDirectory();
+    initializeExecutors();
     return null;
   }
 
+  private void throwPersistErrorIfExists()
+  {
+    if (persistError != null) {
+      throw new RE(persistError, "Error while persisting");
+    }
+  }
+
+  private void initializeExecutors()
+  {
+    log.info("There will be up to[%d] pending persists", maxPendingPersists);

Review comment:
       It is debug now




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] loquisgon commented on a change in pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
loquisgon commented on a change in pull request #11536:
URL: https://github.com/apache/druid/pull/11536#discussion_r686265933



##########
File path: server/src/main/java/org/apache/druid/segment/realtime/appenderator/BatchAppenderator.java
##########
@@ -177,9 +193,53 @@ public Object startJob()
   {
     tuningConfig.getBasePersistDirectory().mkdirs();
     lockBasePersistDirectory();
+    initializeExecutors();
     return null;
   }
 
+  private void throwPersistErrorIfExists()
+  {
+    if (persistError != null) {
+      throw new RE(persistError, "Error while persisting");
+    }
+  }
+
+  private void initializeExecutors()
+  {
+    log.info("There will be up to[%d] pending persists", maxPendingPersists);

Review comment:
       Changed to debug in next commit




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org

[GitHub] [druid] lgtm-com[bot] commented on pull request #11536: Make persists concurrent with adding rows in batch ingestion

Posted by GitBox <gi...@apache.org>.
lgtm-com[bot] commented on pull request #11536:
URL: https://github.com/apache/druid/pull/11536#issuecomment-913840197


   This pull request **introduces 1 alert** when merging 291420ecb85273419203c7a1fa887fe39974ca34 into 60efbb51d0282c1af9c132e7bd37a43e69867622 - [view on LGTM.com](https://lgtm.com/projects/g/apache/druid/rev/pr-66448cca51d1dcc76cf24f3909f35802f584a764)
   
   **new alerts:**
   
   * 1 for Uncontrolled data used in path expression


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org