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 2018/07/30 15:52:52 UTC
[GitHub] gianm closed pull request #6059: CompactionTask: Reject empty
intervals on construction.
gianm closed pull request #6059: CompactionTask: Reject empty intervals on construction.
URL: https://github.com/apache/incubator-druid/pull/6059
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/docs/content/ingestion/tasks.md b/docs/content/ingestion/tasks.md
index 7d63884e2ee..dace2978dd5 100644
--- a/docs/content/ingestion/tasks.md
+++ b/docs/content/ingestion/tasks.md
@@ -279,13 +279,16 @@ An example of compaction task is
This compaction task reads _all segments_ of the interval `2017-01-01/2018-01-01` and results in new segments.
Note that intervals of the input segments are merged into a single interval of `2017-01-01/2018-01-01` no matter what the segmentGranularity was.
-To controll the number of result segments, you can set `targetPartitionSize` or `numShards`. See [indexTuningConfig](#tuningconfig) for more details.
+To control the number of result segments, you can set `targetPartitionSize` or `numShards`. See [indexTuningConfig](#tuningconfig) for more details.
To merge each day's worth of data into separate segments, you can submit multiple `compact` tasks, one for each day. They will run in parallel.
A compaction task internally generates an `index` task spec for performing compaction work with some fixed parameters.
For example, its `firehose` is always the [ingestSegmentSpec](./firehose.html), and `dimensionsSpec` and `metricsSpec`
include all dimensions and metrics of the input segments by default.
+Compaction tasks will exit with a failure status code, without doing anything, if the interval you specify has no
+data segments loaded in it (or if the interval you specify is empty).
+
The output segment can have different metadata from the input segments unless all input segments have the same metadata.
- Dimensions: since Druid supports schema change, the dimensions can be different across segments even if they are a part of the same dataSource.
diff --git a/indexing-service/src/main/java/io/druid/indexing/common/task/CompactionTask.java b/indexing-service/src/main/java/io/druid/indexing/common/task/CompactionTask.java
index 53855574db2..f81999f15f2 100644
--- a/indexing-service/src/main/java/io/druid/indexing/common/task/CompactionTask.java
+++ b/indexing-service/src/main/java/io/druid/indexing/common/task/CompactionTask.java
@@ -49,6 +49,7 @@
import io.druid.indexing.common.task.IndexTask.IndexIngestionSpec;
import io.druid.indexing.common.task.IndexTask.IndexTuningConfig;
import io.druid.indexing.firehose.IngestSegmentFirehoseFactory;
+import io.druid.java.util.common.IAE;
import io.druid.java.util.common.ISE;
import io.druid.java.util.common.JodaUtils;
import io.druid.java.util.common.Pair;
@@ -136,6 +137,10 @@ public CompactionTask(
Preconditions.checkArgument(interval != null || segments != null, "interval or segments should be specified");
Preconditions.checkArgument(interval == null || segments == null, "one of interval and segments should be null");
+ if (interval != null && interval.toDurationMillis() == 0) {
+ throw new IAE("Interval[%s] is empty, must specify a nonempty interval", interval);
+ }
+
this.interval = interval;
this.segments = segments;
this.dimensionsSpec = dimensionsSpec;
@@ -225,7 +230,7 @@ public TaskStatus run(final TaskToolbox toolbox) throws Exception
}
if (indexTaskSpec == null) {
- log.warn("Failed to generate compaction spec");
+ log.warn("Interval[%s] has no segments, nothing to do.", interval);
return TaskStatus.failure(getId());
} else {
final String json = jsonMapper.writerWithDefaultPrettyPrinter().writeValueAsString(indexTaskSpec);
@@ -237,7 +242,7 @@ public TaskStatus run(final TaskToolbox toolbox) throws Exception
/**
* Generate {@link IndexIngestionSpec} from input segments.
-
+ *
* @return null if input segments don't exist. Otherwise, a generated ingestionSpec.
*/
@Nullable
diff --git a/indexing-service/src/main/java/io/druid/indexing/common/task/IndexTask.java b/indexing-service/src/main/java/io/druid/indexing/common/task/IndexTask.java
index a6d3c53060f..5f2cbad32bc 100644
--- a/indexing-service/src/main/java/io/druid/indexing/common/task/IndexTask.java
+++ b/indexing-service/src/main/java/io/druid/indexing/common/task/IndexTask.java
@@ -41,6 +41,7 @@
import io.druid.data.input.Rows;
import io.druid.hll.HyperLogLogCollector;
import io.druid.indexer.IngestionState;
+import io.druid.indexer.TaskStatus;
import io.druid.indexing.appenderator.ActionBasedSegmentAllocator;
import io.druid.indexing.appenderator.ActionBasedUsedSegmentChecker;
import io.druid.indexing.common.IngestionStatsAndErrorsTaskReport;
@@ -48,7 +49,6 @@
import io.druid.indexing.common.TaskLock;
import io.druid.indexing.common.TaskRealtimeMetricsMonitorBuilder;
import io.druid.indexing.common.TaskReport;
-import io.druid.indexer.TaskStatus;
import io.druid.indexing.common.TaskToolbox;
import io.druid.indexing.common.actions.SegmentTransactionalInsertAction;
import io.druid.indexing.common.actions.TaskActionClient;
@@ -190,7 +190,7 @@ public IndexTask(
@JacksonInject RowIngestionMetersFactory rowIngestionMetersFactory
)
{
- this(
+ this(
id,
makeGroupId(ingestionSchema),
taskResource,
@@ -267,6 +267,13 @@ public boolean isReady(TaskActionClient taskActionClient) throws Exception
static boolean isReady(TaskActionClient actionClient, SortedSet<Interval> intervals) throws IOException
{
+ // Sanity check preventing empty intervals (which cannot be locked, and don't make sense anyway).
+ for (Interval interval : intervals) {
+ if (interval.toDurationMillis() == 0) {
+ throw new ISE("Cannot run with empty interval[%s]", interval);
+ }
+ }
+
final List<TaskLock> locks = getTaskLocks(actionClient);
if (locks.size() == 0) {
try {
@@ -394,7 +401,7 @@ public IndexIngestionSpec getIngestionSchema()
}
@Override
- public TaskStatus run(final TaskToolbox toolbox) throws Exception
+ public TaskStatus run(final TaskToolbox toolbox)
{
try {
if (chatHandlerProvider.isPresent()) {
@@ -777,7 +784,7 @@ private ShardSpecs createShardSpecsFromInput(
determinePartitionsMeters.incrementUnparseable();
if (determinePartitionsMeters.getUnparseable() > ingestionSchema.getTuningConfig()
- .getMaxParseExceptions()) {
+ .getMaxParseExceptions()) {
throw new RuntimeException("Max parse exceptions exceeded, terminating task...");
}
}
@@ -917,7 +924,12 @@ private TaskStatus generateAndPublishSegments(
};
try (
- final Appenderator appenderator = newAppenderator(buildSegmentsFireDepartmentMetrics, toolbox, dataSchema, tuningConfig);
+ final Appenderator appenderator = newAppenderator(
+ buildSegmentsFireDepartmentMetrics,
+ toolbox,
+ dataSchema,
+ tuningConfig
+ );
final BatchAppenderatorDriver driver = newDriver(appenderator, toolbox, segmentAllocator);
final Firehose firehose = firehoseFactory.connect(dataSchema.getParser(), firehoseTempDir)
) {
@@ -1288,7 +1300,8 @@ public IndexTuningConfig(
@Deprecated @JsonProperty("reportParseExceptions") @Nullable Boolean reportParseExceptions,
@JsonProperty("publishTimeout") @Nullable Long publishTimeout, // deprecated
@JsonProperty("pushTimeout") @Nullable Long pushTimeout,
- @JsonProperty("segmentWriteOutMediumFactory") @Nullable SegmentWriteOutMediumFactory segmentWriteOutMediumFactory,
+ @JsonProperty("segmentWriteOutMediumFactory") @Nullable
+ SegmentWriteOutMediumFactory segmentWriteOutMediumFactory,
@JsonProperty("logParseExceptions") @Nullable Boolean logParseExceptions,
@JsonProperty("maxParseExceptions") @Nullable Integer maxParseExceptions,
@JsonProperty("maxSavedParseExceptions") @Nullable Integer maxSavedParseExceptions
@@ -1368,12 +1381,16 @@ private IndexTuningConfig(
this.maxParseExceptions = 0;
this.maxSavedParseExceptions = maxSavedParseExceptions == null ? 0 : Math.min(1, maxSavedParseExceptions);
} else {
- this.maxParseExceptions = maxParseExceptions == null ? TuningConfig.DEFAULT_MAX_PARSE_EXCEPTIONS : maxParseExceptions;
+ this.maxParseExceptions = maxParseExceptions == null
+ ? TuningConfig.DEFAULT_MAX_PARSE_EXCEPTIONS
+ : maxParseExceptions;
this.maxSavedParseExceptions = maxSavedParseExceptions == null
- ? TuningConfig.DEFAULT_MAX_SAVED_PARSE_EXCEPTIONS
- : maxSavedParseExceptions;
+ ? TuningConfig.DEFAULT_MAX_SAVED_PARSE_EXCEPTIONS
+ : maxSavedParseExceptions;
}
- this.logParseExceptions = logParseExceptions == null ? TuningConfig.DEFAULT_LOG_PARSE_EXCEPTIONS : logParseExceptions;
+ this.logParseExceptions = logParseExceptions == null
+ ? TuningConfig.DEFAULT_LOG_PARSE_EXCEPTIONS
+ : logParseExceptions;
}
private static Integer initializeTargetPartitionSize(Integer numShards, Integer targetPartitionSize)
diff --git a/indexing-service/src/test/java/io/druid/indexing/common/task/CompactionTaskTest.java b/indexing-service/src/test/java/io/druid/indexing/common/task/CompactionTaskTest.java
index 0300dc8f7f9..a6d029acee5 100644
--- a/indexing-service/src/test/java/io/druid/indexing/common/task/CompactionTaskTest.java
+++ b/indexing-service/src/test/java/io/druid/indexing/common/task/CompactionTaskTest.java
@@ -453,6 +453,28 @@ public void testMissingMetadata() throws IOException, SegmentLoadingException
);
}
+ @Test
+ public void testEmptyInterval() throws Exception
+ {
+ expectedException.expect(IllegalArgumentException.class);
+ expectedException.expectMessage(CoreMatchers.containsString("must specify a nonempty interval"));
+
+ final CompactionTask task = new CompactionTask(
+ null,
+ null,
+ "foo",
+ Intervals.of("2000-01-01/2000-01-01"),
+ null,
+ null,
+ null,
+ null,
+ objectMapper,
+ AuthTestUtils.TEST_AUTHORIZER_MAPPER,
+ new NoopChatHandlerProvider(),
+ null
+ );
+ }
+
private static DimensionsSpec getExpectedDimensionsSpecForAutoGeneration()
{
return new DimensionsSpec(
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
With regards,
Apache Git Services
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org