diff --git a/core/src/main/java/org/apache/druid/common/config/NullValueHandlingConfig.java b/core/src/main/java/org/apache/druid/common/config/NullValueHandlingConfig.java
index ae15d95969c..fa059478226 100644
--- a/core/src/main/java/org/apache/druid/common/config/NullValueHandlingConfig.java
+++ b/core/src/main/java/org/apache/druid/common/config/NullValueHandlingConfig.java
@@ -24,7 +24,7 @@ import com.fasterxml.jackson.annotation.JsonProperty;
public class NullValueHandlingConfig
{
- private static final String NULL_HANDLING_CONFIG_STRING = "druid.generic.useDefaultValueForNull";
+ public static final String NULL_HANDLING_CONFIG_STRING = "druid.generic.useDefaultValueForNull";
@JsonProperty("useDefaultValueForNull")
private final boolean useDefaultValuesForNull;
diff --git a/indexing-service/pom.xml b/indexing-service/pom.xml
index 19397787a85..82564cbacdf 100644
--- a/indexing-service/pom.xml
+++ b/indexing-service/pom.xml
@@ -277,6 +277,11 @@
equalsverifier
test
+
+ com.github.stefanbirkner
+ system-rules
+ test
+
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
index a7c4396b780..aaa31ad134e 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractMultiPhaseParallelIndexingTest.java
@@ -108,7 +108,8 @@ abstract class AbstractMultiPhaseParallelIndexingTest extends AbstractParallelIn
File inputDir,
String filter,
DimensionBasedPartitionsSpec partitionsSpec,
- int maxNumConcurrentSubTasks
+ int maxNumConcurrentSubTasks,
+ TaskState expectedTaskStatus
) throws Exception
{
final ParallelIndexSupervisorTask task = newTask(
@@ -129,7 +130,7 @@ abstract class AbstractMultiPhaseParallelIndexingTest extends AbstractParallelIn
TaskStatus taskStatus = task.run(toolbox);
- Assert.assertEquals(TaskState.SUCCESS, taskStatus.getStatusCode());
+ Assert.assertEquals(expectedTaskStatus, taskStatus.getStatusCode());
shutdownTask(task);
return actionClient.getPublishedSegments();
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
index 029d2bf20c6..9731272fc50 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/AbstractParallelIndexSupervisorTaskTest.java
@@ -170,6 +170,10 @@ public class AbstractParallelIndexSupervisorTaskTest extends IngestionTestBase
catch (EntryExistsException e) {
throw new RuntimeException(e);
}
+
+ // WARNING: In production, subtasks are created via HTTP calls and instantiated by Jackson, which means they
+ // cannot share objects like they can here. For example, if the indexing task uses JsonParseSpec, the same
+ // JSONFlattenerMaker instance is shared among subtasks, which is bad since JSONFlattenerMaker is not thread-safe.
tasks.put(subTask.getId(), service.submit(() -> {
try {
final TaskToolbox toolbox = createTaskToolbox(subTask);
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HashPartitionMultiPhaseParallelIndexingTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HashPartitionMultiPhaseParallelIndexingTest.java
index 7219f16ef88..4670d01b7b8 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HashPartitionMultiPhaseParallelIndexingTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/HashPartitionMultiPhaseParallelIndexingTest.java
@@ -26,6 +26,7 @@ import org.apache.druid.data.input.impl.CSVParseSpec;
import org.apache.druid.data.input.impl.DimensionsSpec;
import org.apache.druid.data.input.impl.ParseSpec;
import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.indexer.TaskState;
import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
import org.apache.druid.indexing.common.LockGranularity;
import org.apache.druid.indexing.common.TaskToolbox;
@@ -131,7 +132,8 @@ public class HashPartitionMultiPhaseParallelIndexingTest extends AbstractMultiPh
inputDir,
"test_*",
new HashedPartitionsSpec(null, 2, ImmutableList.of("dim1", "dim2")),
- MAX_NUM_CONCURRENT_SUB_TASKS
+ MAX_NUM_CONCURRENT_SUB_TASKS,
+ TaskState.SUCCESS
);
assertHashedPartition(publishedSegments);
}
diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/RangePartitionMultiPhaseParallelIndexingTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/RangePartitionMultiPhaseParallelIndexingTest.java
index fcdd5bb088e..a4b83320e57 100644
--- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/RangePartitionMultiPhaseParallelIndexingTest.java
+++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/batch/parallel/RangePartitionMultiPhaseParallelIndexingTest.java
@@ -25,11 +25,13 @@ import com.google.common.collect.ImmutableList;
import com.google.common.collect.Multimap;
import com.google.common.collect.SetMultimap;
import org.apache.druid.client.indexing.IndexingServiceClient;
+import org.apache.druid.common.config.NullValueHandlingConfig;
import org.apache.druid.data.input.InputSplit;
import org.apache.druid.data.input.impl.CSVParseSpec;
import org.apache.druid.data.input.impl.DimensionsSpec;
import org.apache.druid.data.input.impl.ParseSpec;
import org.apache.druid.data.input.impl.TimestampSpec;
+import org.apache.druid.indexer.TaskState;
import org.apache.druid.indexer.partitions.SingleDimensionPartitionsSpec;
import org.apache.druid.indexing.common.LockGranularity;
import org.apache.druid.indexing.common.TaskToolbox;
@@ -38,7 +40,6 @@ import org.apache.druid.indexing.common.task.TaskResource;
import org.apache.druid.indexing.common.task.TestAppenderatorsManager;
import org.apache.druid.java.util.common.ISE;
import org.apache.druid.java.util.common.Intervals;
-import org.apache.druid.java.util.common.StringUtils;
import org.apache.druid.java.util.common.guava.Comparators;
import org.apache.druid.query.scan.ScanResultValue;
import org.apache.druid.timeline.DataSegment;
@@ -48,7 +49,9 @@ import org.hamcrest.Matchers;
import org.joda.time.Interval;
import org.junit.Assert;
import org.junit.Before;
+import org.junit.Rule;
import org.junit.Test;
+import org.junit.contrib.java.lang.system.ProvideSystemProperty;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -67,63 +70,76 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
-import java.util.SortedSet;
-import java.util.TreeSet;
import java.util.stream.Collectors;
@RunWith(Parameterized.class)
public class RangePartitionMultiPhaseParallelIndexingTest extends AbstractMultiPhaseParallelIndexingTest
{
+ private static final boolean USE_INPUT_FORMAT_API = true;
+ private static final boolean USE_MULTIVALUE_DIM = true;
private static final int NUM_FILE = 10;
private static final int NUM_ROW = 20;
- private static final int NUM_DAY = 2;
+ private static final int DIM_FILE_CARDINALITY = 2;
private static final int NUM_PARTITION = 2;
private static final int YEAR = 2017;
+ private static final String TIME = "ts";
private static final String DIM1 = "dim1";
private static final String DIM2 = "dim2";
+ private static final String LIST_DELIMITER = "|";
private static final List DIMS = ImmutableList.of(DIM1, DIM2);
private static final String TEST_FILE_NAME_PREFIX = "test_";
private static final ParseSpec PARSE_SPEC = new CSVParseSpec(
new TimestampSpec(
- "ts",
+ TIME,
"auto",
null
),
new DimensionsSpec(
- DimensionsSpec.getDefaultSchemas(Arrays.asList("ts", DIM1, DIM2)),
- new ArrayList<>(),
- new ArrayList<>()
+ DimensionsSpec.getDefaultSchemas(Arrays.asList(TIME, DIM1, DIM2)),
+ Collections.emptyList(),
+ Collections.emptyList()
),
- null,
- Arrays.asList("ts", DIM1, DIM2, "val"),
+ LIST_DELIMITER,
+ Arrays.asList(TIME, DIM1, DIM2, "val"),
false,
0
);
- @Parameterized.Parameters(name = "{0}, useInputFormatApi={1}, maxNumConcurrentSubTasks={2}")
+ @Parameterized.Parameters(name = "{0}, useInputFormatApi={1}, maxNumConcurrentSubTasks={2}, useMultiValueDim={3}")
public static Iterable