Fix idempotence of segment allocation and task report apis in native batch ingestion by jihoonson · Pull Request #11189 · apache/druid

jihoonson · 2021-05-03T07:19:12Z

Description

Most internal APIs used in Druid's ingestion should be idempotent to handle transient errors. This PR fixes the idempotence of two APIs used in native batch ingestion.

The first API is the segment allocation API used in dynamic partitioning. Currently, transient network errors or task failures can lead to non-contiguous segment partitionIds allocated by this API. This can be a problem because PartitionHolder of those segments of non-contiguous partitionIds will be never complete in the broker timeline. As a result, everything will look fine, that means, the task will succeed, segments will be published into the metadata store, historicals will load and announce them, but you will never be able to query them.

To fix the segment allocation API, I had to add a new API that accepts extra parameters such as sequenceName to guarantee the idempotence. This will break the rolling upgrade that replaces idle nodes with a newer version one at a time. To resolve this issue, I added a taskContext, useLineageBasedSegmentAllocation, to control which protocol to use for segment allocation in dynamic partitioning. This option is true by default and must be set to false during rolling upgrade. The in-place rolling upgrade is not a consideration because batch ingestion doesn't support it.

The second API is the task report API used in all native batch ingestion types. This API can handle retries triggered by transient network errors, but cannot handle duplicate reports by task retries. As a result, if there is a task that failed after sending its report, the supervisor task will count both the report of the failed task and that of the retry task. Because of this bug, the parallel task can incorrectly estimate the cardinality of partition column and the distribution of partition column in hash and range partitioning, respectively.

Finally, to test the fix, I added random task failures and API call retries (emulating transient network failures) in AbstractParallelIndexSupervisorTaskTest. All unit tests extending this class, such as CompactionTaskParallelRunTest, HashPartitionMultiPhaseParallelIndexingTest, SinglePhaseParallelIndexingTest, and RangePartitionMultiPhaseParallelIndexingTest, now run with potential transient task failures and API call retries.

Upgrade path to 0.22:

If you upgrade data nodes first as recommended in https://druid.apache.org/docs/latest/operations/rolling-updates.html#rolling-restart-restore-based, there are nothing to do. All batch tasks with dynamic partitioning should succeed during the replacing rolling upgrade.
If you upgrade overlord before middleManagers, you must set druid.indexer.task.default.context = { "useLineageBasedSegmentAllocation": false } during the upgrade, and restore it to { "useLineageBasedSegmentAllocation": true } after the upgrade is finished.

Key changed/added classes in this PR

SinglePhaseParallelIndexTaskRunner
TaskMonitor
AbstractParallelIndexSupervisorTaskTest

This PR has:

been self-reviewed.
- using the concurrency checklist (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
added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
added integration tests.
been tested in a test Druid cluster.

batch ingestion

jihoonson · 2021-05-03T08:55:58Z

I think we need a cluster-wide configuration corresponding to the new taskContext. I will add it soon.

jihoonson · 2021-05-03T19:26:59Z

~~I added druid.indexer.task.useLineageBasedSegmentAllocation for middleManagers instead of taskContext.~~

jihoonson · 2021-05-03T22:06:34Z

I manually tested the behavior with druid.indexer.task.default.context = { "useLineageBasedSegmentAllocation": true} during rolling upgrade.

clintropolis

overall approach lgtm

clintropolis · 2021-05-03T09:36:06Z

          try {
            final long sleepTime = delay.getMillis();
-            log.debug(
+            log.warn(


clintropolis · 2021-05-03T09:40:05Z

+          throw new ISE(
+              "Can't compact segments of non-consecutive rootPartition range. Missing partitionIds between [%s] and [%s]",
+              curSegment.getEndRootPartitionId(),
+              nextSegment.getStartRootPartitionId()
+          );


clintropolis · 2021-05-06T22:42:42Z

+      final int maxTries,
+      @Nullable final CleanupAfterFailure cleanupAfterFailure,
+      @Nullable final String messageOnRetry,
+      boolean skipSleep


nit: is skip sleep the test parameter i guess? maybe worth javadocs

Added javadocs.

clintropolis · 2021-05-06T22:46:35Z

+|`taskLockTimeout`|300000|task lock timeout in millisecond. For more details, see [Locking](#locking).|
+|`forceTimeChunkLock`|true|_Setting this to false is still experimental_<br/> Force to always use time chunk lock. If not set, each task automatically chooses a lock type to use. If this set, it will overwrite the `druid.indexer.tasklock.forceTimeChunkLock` [configuration for the overlord](../configuration/index.md#overlord-operations). See [Locking](#locking) for more details.|
+|`priority`|Different based on task types. See [Priority](#priority).|Task priority|
+|`useLineageBasedSegmentAllocation`|false|Enable the new lineage-based segment allocation protocol for the native Parallel task with dynamic partitioning. This option should be off during the replacing rolling upgrade to Druid 0.22 or higher. Once the upgrade is done, it must be set to true.|


maybe worth elaborating on why, e.g. "...must be set to true to ensure data correctness"

Suggest also adding a note that this applies if upgrading from a pre-0.22.0 version

Updated the doc per suggestion.

clintropolis · 2021-05-06T23:00:52Z


    // Set forceTimeChunkLock before adding task spec to taskStorage, so that we can see always consistent task spec.
    task.addToContextIfAbsent(Tasks.FORCE_TIME_CHUNK_LOCK_KEY, lockConfig.isForceTimeChunkLock());
+    defaultTaskConfig.getContext().forEach(task::addToContextIfAbsent);


I think we should we also set the use lineage config if it is absent to true here so that custom taskContext configs that are missing that setting do not run with false. The documentation for the default config would then no longer need to indicate that config is the default config since it would be implicit

Good idea. I changed the default of the default context to an empty map, and added useLineageBasedSegmentAllocation here.

clintropolis · 2021-05-06T23:04:17Z

+  /**
+   * Transient task failure rate emulated by the taskKiller in {@link SimpleThreadingTaskRunner}.
+   * Per {@link SubTaskSpec}, there could be at most one task failure.
+   */
+  private final double transientTaskFailureRate;
+
+  /**
+   * Transient API call failure rate emulated by {@link LocalParallelIndexSupervisorTaskClient}.
+   * This will be applied to every API calls in the future.
+   */
+  private final double transientApiCallFailureRate;
+


clintropolis · 2021-05-06T23:10:58Z

+              "Cannot publish segments due to incomplete time chunk. Segments are [%s]",
+              segmentsPerInterval.stream().map(DataSegment::getId).collect(Collectors.toList())


👍 on sanity check... is there any chance the list of segments is huge here? (maybe we should use log.errorSegments to log segments and just include count/interval or something?)

Thanks for reminding me of that. Changed to use log.errorSegments and to not create a too large string for exception.

clintropolis

lgtm 🤘

jihoonson · 2021-05-07T21:29:42Z

@jon-wei @clintropolis thanks for the review!

eeren0 · 2021-06-22T10:08:36Z

Hi, I am wondering if this fix may be related to the issue observed in #11348 as well?

… native batch ingestion (apache#11189)" This reverts commit 2df4214.

jihoonson added 2 commits May 2, 2021 23:30

Fix idempotence of segment allocation and task report apis in native

7e088d2

batch ingestion

better error and javadoc

72b6e36

jihoonson added Bug Area - Batch Ingestion Incompatible Release Notes labels May 3, 2021

jihoonson added 4 commits May 3, 2021 09:29

checkstyle and dependency

f6cefa3

fix tests and add more tests

d439ddc

task config instead of context; add doc

0adf83a

unused import and dependency

f354e38

jihoonson added 4 commits May 3, 2021 13:25

typo in doc

95d0d6b

fix unintended changes

76bc3b9

fix wrong import

7501484

remove unnecessary error handling

123dc9b

jihoonson added 3 commits May 3, 2021 18:13

add task context back

59d8d3c

default task context

5f2a19e

fix test and doc

b016a91

clintropolis reviewed May 6, 2021

View reviewed changes

address comments

5d1c5b6

jon-wei approved these changes May 7, 2021

View reviewed changes

unused imports

c08877a

clintropolis approved these changes May 7, 2021

View reviewed changes

jihoonson merged commit 2df4214 into apache:master May 7, 2021

abhishekagarwal87 mentioned this pull request Jun 2, 2021

Incomplete numbered shard specs partition when index task fails and retries #11322

Closed

maytasm mentioned this pull request Aug 5, 2021

Fix ingestion task failure when no input split to process #11553

Merged

9 tasks

clintropolis added this to the 0.22.0 milestone Aug 12, 2021

clintropolis mentioned this pull request Sep 3, 2021

[Draft] 0.22.0 Release Notes #11657

Closed

jon-wei pushed a commit to jon-wei/druid that referenced this pull request Nov 22, 2021

Revert "Fix idempotence of segment allocation and task report apis in…

d1a9a1a

… native batch ingestion (apache#11189)" This reverts commit 2df4214.

		"Cannot publish segments due to incomplete time chunk. Segments are [%s]",
		segmentsPerInterval.stream().map(DataSegment::getId).collect(Collectors.toList())

Conversation

jihoonson commented May 3, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

Key changed/added classes in this PR

Uh oh!

jihoonson commented May 3, 2021

Uh oh!

jihoonson commented May 3, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

jihoonson commented May 3, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

clintropolis left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

clintropolis left a comment

Choose a reason for hiding this comment

Uh oh!

jihoonson commented May 7, 2021

Uh oh!

eeren0 commented Jun 22, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

jihoonson commented May 3, 2021 •

edited

Loading

jihoonson commented May 3, 2021 •

edited

Loading

jihoonson commented May 3, 2021 •

edited

Loading