Skip to content

Allow client to configure batch ingestion task to wait to complete until segments are confirmed to be available by other #10676

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
suneet-s merged 36 commits into from
Apr 9, 2021
Merged

Allow client to configure batch ingestion task to wait to complete until segments are confirmed to be available by other #10676

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
36 commits
Select commit Hold shift + click to select a range
d9e7f9b
Add ability to wait for segment availability for batch jobs
Dec 9, 2020
3d4de8c
IT updates
Dec 11, 2020
027938e
fix queries in legacy hadoop IT
Dec 11, 2020
8b9d26d
Fix broken indexing integration tests
Dec 14, 2020
72cd38a
address an lgtm flag
Dec 15, 2020
f47a8bd
spell checker still flagging for hadoop doc. adding under that file h…
Dec 15, 2020
d5ed3c8
fix compaction IT
Dec 15, 2020
71d6f18
Updates to wait for availability method
Dec 15, 2020
0170dca
improve unit testing for patch
Dec 15, 2020
237325f
fix bad indentation
Dec 17, 2020
848805e
refactor waitForSegmentAvailability
Dec 17, 2020
8169c88
Merge branch 'master' into batch-ingest-wait-for-handoff
Dec 28, 2020
32fdff5
Fixes based off of review comments
Jan 5, 2021
b4f2e09
Merge branch 'master' into batch-ingest-wait-for-handoff
Jan 6, 2021
9295df1
cleanup to get compile after merging with master
Jan 6, 2021
abf7f99
fix failing test after previous logic update
Jan 6, 2021
d2e9918
add back code that must have gotten deleted during conflict resolution
Jan 6, 2021
92f3bec
Merge branch 'master' into batch-ingest-wait-for-handoff
Jan 11, 2021
222d3c2
update some logging code
Jan 28, 2021
88df46c
Merge branch 'master' into batch-ingest-wait-for-handoff
Jan 28, 2021
452b649
fixes to get compilation working after merge with master
Jan 28, 2021
376cfa0
reset interrupt flag in catch block after code review pointed it out
Jan 28, 2021
c63611a
small changes following self-review
Jan 28, 2021
1772199
Merge branch 'master' into batch-ingest-wait-for-handoff
Feb 10, 2021
16274c3
Merge branch 'master' into batch-ingest-wait-for-handoff
Mar 17, 2021
503bdb4
fixup some issues brought on by merge with master
Mar 17, 2021
0fa4f41
small changes after review
Mar 25, 2021
3674836
Merge branch 'master' into batch-ingest-wait-for-handoff
Mar 25, 2021
8c3a499
cleanup a little bit after merge with master
Mar 25, 2021
f6c9168
Fix potential resource leak in AbstractBatchIndexTask
Mar 30, 2021
edc26dd
syntax fix
Mar 30, 2021
7b25321
Add a Compcation TuningConfig type
Mar 31, 2021
9498cb0
add docs stipulating the lack of support by Compaction tasks for the …
Mar 31, 2021
d39dfa3
Merge branch 'master' into batch-ingest-wait-for-handoff
Apr 6, 2021
747ee33
Fixup compilation errors after merge with master
Apr 6, 2021
381cdfe
Remove erreneous newline
Apr 7, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions core/src/main/java/org/apache/druid/indexer/IngestionState.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,5 +24,6 @@ public enum IngestionState
NOT_STARTED,
DETERMINE_PARTITIONS,
BUILD_SEGMENTS,
SEGMENT_AVAILABILITY_WAIT,
COMPLETED
}
2 changes: 1 addition & 1 deletion docs/ingestion/compaction.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -111,7 +111,7 @@ To perform a manual compaction, you submit a compaction task. Compaction tasks m
|`dimensionsSpec`|Custom dimensions spec. The compaction task uses the specified dimensions spec if it exists instead of generating one.|No|
|`metricsSpec`|Custom metrics spec. The compaction task uses the specified metrics spec rather than generating one.|No|
|`segmentGranularity`|When set, the compaction task changes the segment granularity for the given interval. Deprecated. Use `granularitySpec`. |No.|
|`tuningConfig`|[Parallel indexing task tuningConfig](native-batch.md#tuningconfig)|No|
|`tuningConfig`|[Parallel indexing task tuningConfig](native-batch.md#tuningconfig). Note that your tuning config cannot contain a non-zero value for `awaitSegmentAvailabilityTimeoutMillis` because it is not supported by compaction tasks at this time.|No|
|`context`|[Task context](./tasks.md#context)|No|
|`granularitySpec`|Custom `granularitySpec` to describe the `segmentGranularity` and `queryGranularity` for the compacted segments. See [Compaction granularitySpec](#compaction-granularity-spec).|No|

Expand Down
1 change: 1 addition & 0 deletions docs/ingestion/hadoop.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -336,6 +336,7 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
|logParseExceptions|Boolean|If true, log an error message when a parsing exception occurs, containing information about the row where the error occurred.|no(default = false)|
|maxParseExceptions|Integer|The maximum number of parse exceptions that can occur before the task halts ingestion and fails. Overrides `ignoreInvalidRows` if `maxParseExceptions` is defined.|no(default = unlimited)|
|useYarnRMJobStatusFallback|Boolean|If the Hadoop jobs created by the indexing task are unable to retrieve their completion status from the JobHistory server, and this parameter is true, the indexing task will try to fetch the application status from `http://<yarn-rm-address>/ws/v1/cluster/apps/<application-id>`, where `<yarn-rm-address>` is the value of `yarn.resourcemanager.webapp.address` in your Hadoop configuration. This flag is intended as a fallback for cases where an indexing task's jobs succeed, but the JobHistory server is unavailable, causing the indexing task to fail because it cannot determine the job statuses.|no (default = true)|
|awaitSegmentAvailabilityTimeoutMillis|Long|Milliseconds to wait for the newly indexed segments to become available for query after ingestion completes. If `<= 0`, no wait will occur. If `> 0`, the task will wait for the Coordinator to indicate that the new segments are available for querying. If the timeout expires, the task will exit as successful, but the segments were not confirmed to have become available for query.|no (default = 0)|

### `jobProperties`

Expand Down
2 changes: 2 additions & 0 deletions docs/ingestion/native-batch.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -197,6 +197,7 @@ A sample task is shown below:
|id|The task ID. If this is not explicitly specified, Druid generates the task ID using task type, data source name, interval, and date-time stamp. |no|
|spec|The ingestion spec including the data schema, IOConfig, and TuningConfig. See below for more details. |yes|
|context|Context containing various task configuration parameters. See below for more details.|no|
|awaitSegmentAvailabilityTimeoutMillis|Long|Milliseconds to wait for the newly indexed segments to become available for query after ingestion completes. If `<= 0`, no wait will occur. If `> 0`, the task will wait for the Coordinator to indicate that the new segments are available for querying. If the timeout expires, the task will exit as successful, but the segments were not confirmed to have become available for query. Note for compaction tasks: you should not set this to a non-zero value because it is not supported by the compaction task type at this time.|no (default = 0)|

### `dataSchema`

Expand Down Expand Up @@ -250,6 +251,7 @@ The tuningConfig is optional and default parameters will be used if no tuningCon
|taskStatusCheckPeriodMs|Polling period in milliseconds to check running task statuses.|1000|no|
|chatHandlerTimeout|Timeout for reporting the pushed segments in worker tasks.|PT10S|no|
|chatHandlerNumRetries|Retries for reporting the pushed segments in worker tasks.|5|no|
|awaitSegmentAvailabilityTimeoutMillis|Long|Milliseconds to wait for the newly indexed segments to become available for query after ingestion completes. If `<= 0`, no wait will occur. If `> 0`, the task will wait for the Coordinator to indicate that the new segments are available for querying. If the timeout expires, the task will exit as successful, but the segments were not confirmed to have become available for query.|no (default = 0)|

### Split Hint Spec

Expand Down
3 changes: 2 additions & 1 deletion .../main/java/org/apache/druid/indexing/materializedview/MaterializedViewSupervisorSpec.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,7 +199,8 @@ public HadoopIndexTask createTask(Interval interval, String version, List<DataSe
tuningConfig.getUserAllowedHadoopPrefix(),
tuningConfig.isLogParseExceptions(),
tuningConfig.getMaxParseExceptions(),
tuningConfig.isUseYarnRMJobStatusFallback()
tuningConfig.isUseYarnRMJobStatusFallback(),
tuningConfig.getAwaitSegmentAvailabilityTimeoutMillis()
);

// generate granularity
Expand Down
4 changes: 2 additions & 2 deletions ...ka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/KafkaIndexTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -125,6 +125,8 @@
import org.apache.druid.query.timeseries.TimeseriesQueryEngine;
import org.apache.druid.query.timeseries.TimeseriesQueryQueryToolChest;
import org.apache.druid.query.timeseries.TimeseriesQueryRunnerFactory;
import org.apache.druid.segment.handoff.SegmentHandoffNotifier;
import org.apache.druid.segment.handoff.SegmentHandoffNotifierFactory;
import org.apache.druid.segment.incremental.RowIngestionMeters;
import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
import org.apache.druid.segment.incremental.RowIngestionMetersTotals;
Expand All @@ -136,8 +138,6 @@
import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
import org.apache.druid.segment.realtime.firehose.NoopChatHandlerProvider;
import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifier;
import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
import org.apache.druid.segment.transform.ExpressionTransform;
import org.apache.druid.segment.transform.TransformSpec;
import org.apache.druid.server.DruidNode;
Expand Down
4 changes: 2 additions & 2 deletions ...ndexing-service/src/test/java/org/apache/druid/indexing/kinesis/KinesisIndexTaskTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,15 +97,15 @@
import org.apache.druid.query.timeseries.TimeseriesQueryQueryToolChest;
import org.apache.druid.query.timeseries.TimeseriesQueryRunnerFactory;
import org.apache.druid.segment.TestHelper;
import org.apache.druid.segment.handoff.SegmentHandoffNotifier;
import org.apache.druid.segment.handoff.SegmentHandoffNotifierFactory;
import org.apache.druid.segment.incremental.RowIngestionMeters;
import org.apache.druid.segment.indexing.DataSchema;
import org.apache.druid.segment.join.NoopJoinableFactory;
import org.apache.druid.segment.loading.DataSegmentPusher;
import org.apache.druid.segment.loading.LocalDataSegmentPusher;
import org.apache.druid.segment.loading.LocalDataSegmentPusherConfig;
import org.apache.druid.segment.realtime.firehose.NoopChatHandlerProvider;
import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifier;
import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
import org.apache.druid.segment.transform.ExpressionTransform;
import org.apache.druid.segment.transform.TransformSpec;
import org.apache.druid.server.DruidNode;
Expand Down
26 changes: 22 additions & 4 deletions indexing-hadoop/src/main/java/org/apache/druid/indexer/HadoopTuningConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,7 @@ public static HadoopTuningConfig makeDefaultTuningConfig()
null,
null,
null,
null,
null
);
}
Expand Down Expand Up @@ -101,6 +102,7 @@ public static HadoopTuningConfig makeDefaultTuningConfig()
private final boolean logParseExceptions;
private final int maxParseExceptions;
private final boolean useYarnRMJobStatusFallback;
private final long awaitSegmentAvailabilityTimeoutMillis;

@JsonCreator
public HadoopTuningConfig(
Expand Down Expand Up @@ -130,7 +132,8 @@ public HadoopTuningConfig(
final @JsonProperty("allowedHadoopPrefix") @Nullable List<String> allowedHadoopPrefix,
final @JsonProperty("logParseExceptions") @Nullable Boolean logParseExceptions,
final @JsonProperty("maxParseExceptions") @Nullable Integer maxParseExceptions,
final @JsonProperty("useYarnRMJobStatusFallback") @Nullable Boolean useYarnRMJobStatusFallback
final @JsonProperty("useYarnRMJobStatusFallback") @Nullable Boolean useYarnRMJobStatusFallback,
final @JsonProperty("awaitSegmentAvailabilityTimeoutMillis") @Nullable Long awaitSegmentAvailabilityTimeoutMillis
)
{
this.workingPath = workingPath;
Expand Down Expand Up @@ -176,6 +179,12 @@ public HadoopTuningConfig(
this.logParseExceptions = logParseExceptions == null ? TuningConfig.DEFAULT_LOG_PARSE_EXCEPTIONS : logParseExceptions;

this.useYarnRMJobStatusFallback = useYarnRMJobStatusFallback == null ? true : useYarnRMJobStatusFallback;

if (awaitSegmentAvailabilityTimeoutMillis == null || awaitSegmentAvailabilityTimeoutMillis < 0) {
this.awaitSegmentAvailabilityTimeoutMillis = DEFAULT_AWAIT_SEGMENT_AVAILABILITY_TIMEOUT_MILLIS;
} else {
this.awaitSegmentAvailabilityTimeoutMillis = awaitSegmentAvailabilityTimeoutMillis;
}
}

@Nullable
Expand Down Expand Up @@ -334,6 +343,12 @@ public boolean isUseYarnRMJobStatusFallback()
return useYarnRMJobStatusFallback;
}

@JsonProperty
public long getAwaitSegmentAvailabilityTimeoutMillis()
{
return awaitSegmentAvailabilityTimeoutMillis;
}

public HadoopTuningConfig withWorkingPath(String path)
{
return new HadoopTuningConfig(
Expand Down Expand Up @@ -361,7 +376,8 @@ public HadoopTuningConfig withWorkingPath(String path)
allowedHadoopPrefix,
logParseExceptions,
maxParseExceptions,
useYarnRMJobStatusFallback
useYarnRMJobStatusFallback,
awaitSegmentAvailabilityTimeoutMillis
);
}

Expand Down Expand Up @@ -392,7 +408,8 @@ public HadoopTuningConfig withVersion(String ver)
allowedHadoopPrefix,
logParseExceptions,
maxParseExceptions,
useYarnRMJobStatusFallback
useYarnRMJobStatusFallback,
awaitSegmentAvailabilityTimeoutMillis
);
}

Expand Down Expand Up @@ -423,7 +440,8 @@ public HadoopTuningConfig withShardSpecs(Map<Long, List<HadoopyShardSpec>> specs
allowedHadoopPrefix,
logParseExceptions,
maxParseExceptions,
useYarnRMJobStatusFallback
useYarnRMJobStatusFallback,
awaitSegmentAvailabilityTimeoutMillis
);
}
}
1 change: 1 addition & 0 deletions indexing-hadoop/src/test/java/org/apache/druid/indexer/BatchDeltaIngestionTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -481,6 +481,7 @@ private HadoopDruidIndexerConfig makeHadoopDruidIndexerConfig(
null,
null,
null,
null,
null
)
)
Expand Down
1 change: 1 addition & 0 deletions indexing-hadoop/src/test/java/org/apache/druid/indexer/DetermineHashedPartitionsJobTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -235,6 +235,7 @@ public DetermineHashedPartitionsJobTest(
null,
null,
null,
null,
null
)
);
Expand Down
1 change: 1 addition & 0 deletions indexing-hadoop/src/test/java/org/apache/druid/indexer/DeterminePartitionsJobTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -344,6 +344,7 @@ public DeterminePartitionsJobTest(
null,
null,
null,
null,
null
)
)
Expand Down
1 change: 1 addition & 0 deletions indexing-hadoop/src/test/java/org/apache/druid/indexer/HadoopDruidIndexerConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -257,6 +257,7 @@ HadoopIngestionSpec build()
null,
null,
null,
null,
null
);

Expand Down
2 changes: 2 additions & 0 deletions indexing-hadoop/src/test/java/org/apache/druid/indexer/HadoopTuningConfigTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,7 @@ public void testSerde() throws Exception
null,
null,
null,
null,
null
);

Expand All @@ -86,6 +87,7 @@ public void testSerde() throws Exception
Assert.assertEquals(0, actual.getNumBackgroundPersistThreads());
Assert.assertEquals(true, actual.isForceExtendableShardSpecs());
Assert.assertEquals(true, actual.isUseExplicitVersion());
Assert.assertEquals(0, actual.getAwaitSegmentAvailabilityTimeoutMillis());
}

public static <T> T jsonReadWriteRead(String s, Class<T> klass)
Expand Down
1 change: 1 addition & 0 deletions indexing-hadoop/src/test/java/org/apache/druid/indexer/IndexGeneratorJobTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -547,6 +547,7 @@ public void setUp() throws Exception
null,
null,
null,
null,
null
)
)
Expand Down
1 change: 1 addition & 0 deletions indexing-hadoop/src/test/java/org/apache/druid/indexer/JobHelperTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -187,6 +187,7 @@ public void setup() throws Exception
null,
null,
null,
null,
null
)
)
Expand Down
1 change: 1 addition & 0 deletions indexing-hadoop/src/test/java/org/apache/druid/indexer/path/GranularityPathSpecTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@ public class GranularityPathSpecTest
null,
null,
null,
null,
null
);

Expand Down
4 changes: 3 additions & 1 deletion indexing-service/src/main/java/org/apache/druid/guice/IndexingServiceTuningConfigModule.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.google.common.collect.ImmutableList;
import com.google.inject.Binder;
import org.apache.druid.indexing.common.task.CompactionTask;
import org.apache.druid.indexing.common.task.IndexTask.IndexTuningConfig;
import org.apache.druid.indexing.common.task.batch.parallel.ParallelIndexTuningConfig;
import org.apache.druid.initialization.DruidModule;
Expand All @@ -39,7 +40,8 @@ public List<? extends Module> getJacksonModules()
new SimpleModule(IndexingServiceTuningConfigModule.class.getSimpleName())
.registerSubtypes(
new NamedType(IndexTuningConfig.class, "index"),
new NamedType(ParallelIndexTuningConfig.class, "index_parallel")
new NamedType(ParallelIndexTuningConfig.class, "index_parallel"),
new NamedType(CompactionTask.CompactionTuningConfig.class, "compaction")
)
);
}
Expand Down
25 changes: 22 additions & 3 deletions ...src/main/java/org/apache/druid/indexing/common/IngestionStatsAndErrorsTaskReportData.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,17 +41,22 @@ public class IngestionStatsAndErrorsTaskReportData
@Nullable
private String errorMsg;

@JsonProperty
private boolean segmentAvailabilityConfirmed;

public IngestionStatsAndErrorsTaskReportData(
@JsonProperty("ingestionState") IngestionState ingestionState,
@JsonProperty("unparseableEvents") Map<String, Object> unparseableEvents,
@JsonProperty("rowStats") Map<String, Object> rowStats,
@JsonProperty("errorMsg") @Nullable String errorMsg
@JsonProperty("errorMsg") @Nullable String errorMsg,
@JsonProperty("segmentAvailabilityConfirmed") boolean segmentAvailabilityConfirmed
Copy link
Copy Markdown
Contributor

@jihoonson jihoonson Jan 5, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can you give some more details on how this will be used in your application? Do you want to track handoff failures of each task? I'm wondering if handoff time is also important.

Copy link
Copy Markdown
Contributor Author

@capistrant capistrant Jan 5, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

my company deploys a large multi-tenant cluster with a services layer for ingestion that our tenants use. these tenants don't just want to know when their task succeeds, they also want to know when data from batch ingest is available for querying. This solution allows us to prevent the ingestion services layer and/or individual tenants from banging on Druid APIs trying to see if their data is available after ingestion.

Copy link
Copy Markdown
Contributor

@jihoonson jihoonson Jan 27, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

my company deploys a large multi-tenant cluster with a services layer for ingestion that our tenants use. these tenants don't just want to know when their task succeeds, they also want to know when data from batch ingest is available for querying. This solution allows us to prevent the ingestion services layer and/or individual tenants from banging on Druid APIs trying to see if their data is available after ingestion.

I understand this, but my question is more like what people expect when segment handoff fails. In streaming ingestion, the handoff failure causes task failure (this behavior seems arguable, but that's what it does now) and thus people's expectation is that they could see some data dropped after handoff failures until new tasks read the same data and publishes the same segments again. However, since there is no realtime querying in batch ingestion, I don't think tasks should fail on handoff failures (which is what this PR does! 🙂), but then what will be people's expectation? Are they going to be just OK with handoff failures and wait indefinitely until historicals load new segments (the current behavior)? Do they want to know why the handoff failed? Do they want to know how long it took before the handoff failed? These questions are not clear to me.

Copy link
Copy Markdown
Contributor Author

@capistrant capistrant Jan 27, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

my company deploys a large multi-tenant cluster with a services layer for ingestion that our tenants use. these tenants don't just want to know when their task succeeds, they also want to know when data from batch ingest is available for querying. This solution allows us to prevent the ingestion services layer and/or individual tenants from banging on Druid APIs trying to see if their data is available after ingestion.

I understand this, but my question is more like what people expect when segment handoff fails. In streaming ingestion, the handoff failure causes task failure (this behavior seems arguable, but that's what it does now) and thus people's expectation is that they could see some data dropped after handoff failures until new tasks read the same data and publishes the same segments again. However, since there is no realtime querying in batch ingestion, I don't think tasks should fail on handoff failures (which is what this PR does! 🙂), but then what will be people's expectation? Are they going to be just OK with handoff failures and wait indefinitely until historicals load new segments (the current behavior)? Do they want to know why the handoff failed? Do they want to know how long it took before the handoff failed? These questions are not clear to me.

good question. for my specific case the service that end users interact with really wanted to be able to answer this question for the end user:

  • Is the data that I ingested in this job completely loaded for querying?

For us a simple yes/no will suffice. The cluster operators would have the goal of having 100% of jobs successfully handoff data before the timeout, but when that doesn't happen our users simply want to know that they may need to wait longer. We are simply trying to be transparent and report the point in time status. The onus of finding out when the data is fully loaded if this timeout expired before loading, would fall on a different solution (TBD).

You're right, we intentionally did not fail these tasks because Historical nodes loading the segments is detached from whether or not the data was written to deepstore/metastore (if that failed the task should and likely would fail due to existing code paths). We don't want our end users thinking they need to re-run their jobs when this is much more likely to be an issue of the coordinator not having assigned servers to load segments by the time the timeout expired.

Why the handoff failed would be something I as an operator am more interested compared to a user (unless that user is also an operator). I think that would be very difficult to communicate in these reports since the indexing task doesn't know much about what the rest of the cluster is doing.

Knowing how long it took before the time out could be found in the spec, but I guess it could be useful to add that value to the report as well if you think users would want to have quick reference. I think that rather than having that static value, it could be cool to have the dynamic time waited for handoff. Maybe it is the static value because we hit the timeout. but as an operator I would enjoy seeing how long each successful job waited for handoff. what do you think about that?

Copy link
Copy Markdown
Contributor

@jihoonson jihoonson Jan 28, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For us a simple yes/no will suffice. The cluster operators would have the goal of having 100% of jobs successfully handoff data before the timeout, but when that doesn't happen our users simply want to know that they may need to wait longer. We are simply trying to be transparent and report the point in time status. The onus of finding out when the data is fully loaded if this timeout expired before loading, would fall on a different solution (TBD).

Cool, are you working on "the different solution"? That would be interesting too.

Why the handoff failed would be something I as an operator am more interested compared to a user (unless that user is also an operator). I think that would be very difficult to communicate in these reports since the indexing task doesn't know much about what the rest of the cluster is doing.

I agree. I think we need more visibility on the coordinator behavior.

Knowing how long it took before the time out could be found in the spec, but I guess it could be useful to add that value to the report as well if you think users would want to have quick reference. I think that rather than having that static value, it could be cool to have the dynamic time waited for handoff. Maybe it is the static value because we hit the timeout. but as an operator I would enjoy seeing how long each successful job waited for handoff. what do you think about that?

That seems useful to me too 👍

For the time to fail handoff, due to the above issue of the lack of ability to know the cause of handoff failures, I guess I was wondering if the report can be a false alarm. For example, the report can say it failed to confirm the segments handed off, but maybe the handoff could be even not triggered at all for some reason. I don't think this can happen for now, but is possible in the future if someone else modifies this area for some good reason. segmentAvailabilityConfirmationCompleted + time to fail handoff can be an indicator of such unexpected failures. I would say this is not a blocker for this PR, but it seems useful to me.

Copy link
Copy Markdown
Contributor

@suneet-s suneet-s Mar 25, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Naive question: Should this be

Suggested change
@JsonProperty("segmentAvailabilityConfirmed") boolean segmentAvailabilityConfirmed
@JsonProperty("segmentAvailabilityConfirmed") @Nullable Boolean segmentAvailabilityConfirmed

Seeing the json properties automatically make me think about version mismatches - but I don't exactly know how this is used - so I'm just asking in the hope you can save me some time from digging :)

Copy link
Copy Markdown
Contributor Author

@capistrant capistrant Mar 25, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

So these will be created/written by individual indexing tasks. then stored wherever the cluster stores task logs. And I believe the only way they are ever accessed by Druid is streamed from their location directly to an API caller without ever deserializing them. So I don't think there is any possibility for issues during an upgrade here.

)
{
this.ingestionState = ingestionState;
this.unparseableEvents = unparseableEvents;
this.rowStats = rowStats;
this.errorMsg = errorMsg;
this.segmentAvailabilityConfirmed = segmentAvailabilityConfirmed;
}

@JsonProperty
Expand Down Expand Up @@ -79,6 +84,12 @@ public String getErrorMsg()
return errorMsg;
}

@JsonProperty
public boolean isSegmentAvailabilityConfirmed()
{
return segmentAvailabilityConfirmed;
}

public static IngestionStatsAndErrorsTaskReportData getPayloadFromTaskReports(
Map<String, TaskReport> taskReports
)
Expand All @@ -100,13 +111,20 @@ public boolean equals(Object o)
return getIngestionState() == that.getIngestionState() &&
Objects.equals(getUnparseableEvents(), that.getUnparseableEvents()) &&
Objects.equals(getRowStats(), that.getRowStats()) &&
Objects.equals(getErrorMsg(), that.getErrorMsg());
Objects.equals(getErrorMsg(), that.getErrorMsg()) &&
Objects.equals(isSegmentAvailabilityConfirmed(), that.isSegmentAvailabilityConfirmed());
}

@Override
public int hashCode()
{
return Objects.hash(getIngestionState(), getUnparseableEvents(), getRowStats(), getErrorMsg());
return Objects.hash(
getIngestionState(),
getUnparseableEvents(),
getRowStats(),
getErrorMsg(),
isSegmentAvailabilityConfirmed()
);
}

@Override
Expand All @@ -117,6 +135,7 @@ public String toString()
", unparseableEvents=" + unparseableEvents +
", rowStats=" + rowStats +
", errorMsg='" + errorMsg + '\'' +
", segmentAvailabilityConfirmed=" + segmentAvailabilityConfirmed +
'}';
}
}
2 changes: 1 addition & 1 deletion indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolbox.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,6 +49,7 @@
import org.apache.druid.query.QueryRunnerFactoryConglomerate;
import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.IndexMergerV9;
import org.apache.druid.segment.handoff.SegmentHandoffNotifierFactory;
import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
import org.apache.druid.segment.join.JoinableFactory;
import org.apache.druid.segment.loading.DataSegmentArchiver;
Expand All @@ -59,7 +60,6 @@
import org.apache.druid.segment.loading.SegmentLoadingException;
import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
import org.apache.druid.segment.realtime.firehose.ChatHandlerProvider;
import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordination.DataSegmentAnnouncer;
import org.apache.druid.server.coordination.DataSegmentServerAnnouncer;
Expand Down
2 changes: 1 addition & 1 deletion indexing-service/src/main/java/org/apache/druid/indexing/common/TaskToolboxFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
import org.apache.druid.query.QueryRunnerFactoryConglomerate;
import org.apache.druid.segment.IndexIO;
import org.apache.druid.segment.IndexMergerV9;
import org.apache.druid.segment.handoff.SegmentHandoffNotifierFactory;
import org.apache.druid.segment.incremental.RowIngestionMetersFactory;
import org.apache.druid.segment.join.JoinableFactory;
import org.apache.druid.segment.loading.DataSegmentArchiver;
Expand All @@ -55,7 +56,6 @@
import org.apache.druid.segment.loading.DataSegmentPusher;
import org.apache.druid.segment.realtime.appenderator.AppenderatorsManager;
import org.apache.druid.segment.realtime.firehose.ChatHandlerProvider;
import org.apache.druid.segment.realtime.plumber.SegmentHandoffNotifierFactory;
import org.apache.druid.server.DruidNode;
import org.apache.druid.server.coordination.DataSegmentAnnouncer;
import org.apache.druid.server.coordination.DataSegmentServerAnnouncer;
Expand Down
Loading