IngestSegmentFirehoseFactory race between tasks

[gianm]

IngestSegmentFirehoseFactory uses a fixed task id "reindex" when creating the toolbox factory that it uses to get segments. This creates a race condition where two tasks could actually download the same segment at the same time into the same directory, and one will get clobbered and fail.

One situation where this can happen easily is if you're reindexing a datasource into two datasources (maybe reducing to two different levels of grain). The tasks will proceed simultaneously, since there's no cross locking, but they'll be downloading the same segments for the input datasource.

One fix is having it actually use the real task id (which would need to get plumbed in). This has the advantage of putting the task files all together, and using the existing mechanisms for cleaning up task work directories.

[zhihuij]

Hi, @gianm,

Do you mean "reingest" here？ because I only see a NoopTask with task id "reingest" in IngestSegmentFirehoseFactory.

Can you give some clue on

using the existing mechanisms for cleaning up task work directories ?

Because I want to fix #3655 , but didn't find a simple way to do it (fixed with a callback currently).

Thank you very much.

[gianm]

@onlychoice, I do mean "reingest". What's going on is that the IngestSegmentFirehoseFactory is creating a TaskToolboxFactory with a dummy NoopTask that has id "reingest". That dummy task id is used by the toolbox as part of its tmp directory. This causes a couple of problems:

• Files left on disk when batch task with IngestSegmentFirehose is done #3655: the tmp directory built from "reingest" isn't cleaned up
• this issue: the tmp directory is shared between tasks and can cause conflicts

Ideally, instead of using an injector to create its own TaskToolboxFactory, and then building a toolbox with a dummy id, the firehose should use the real TaskToolbox with the real task id. This is available at the task level and the task would have to give it to the firehose somehow. I haven't thought of a really nice way to do it, but one gross way would be for tasks to check firehoseFactory instanceof IngestSegmentFirehoseFactory and if so, call some special method to pass in the toolbox.

[zhihuij]

Thanks @gianm , I got your idea, each task should have its own task work directory, and should be cleaned up after task is done.

I have searched through Task, TaskToolbox, TaskRunner, ExecutorLifecycle, but didn't find the code which do the job

the existing mechanisms for cleaning up task work directories

Can you give some clues on this? Thanks.

[zhihuij]

the existing mechanisms for cleaning up task work directories

https://github.com/druid-io/druid/blob/master/indexing-service/src/main/java/io/druid/indexing/overlord/ForkingTaskRunner.java#L492

[gianm]

Yes, that's it, sorry for not getting back to you earlier.