QiuMM opened a new issue #6722: tasks failed because of no sink for identifier URL: https://github.com/apache/incubator-druid/issues/6722 Recently, I observed some kafka index tasks failed, below is the log: ``` 2018-12-09 20:33:15,603 INFO [task-runner-0-priority-0] io.druid.segment.realtime.appenderator.BaseAppenderatorDriver - Pushing segments in background: [aweme_low_frequency_all_event_2018-12-08T01:00:00.000Z_2018-12-08T02:00:00.000Z_2018-12-09T16:40:28.781Z_58, aweme_low_frequency_all_event_2018-12-09T20:00:00.000Z_2018-12-09T21:00:00.000Z_2018-12-09T20:00:01.734Z_355] 2018-12-09 20:33:15,604 INFO [appenderator_persist_0] io.druid.segment.realtime.appenderator.AppenderatorImpl - Removing sink for segment[aweme_low_frequency_all_event_2018-12-08T01:00:00.000Z_2018-12-08T02:00:00.000Z_2018-12-09T16:40:28.781Z_54]. 2018-12-09 20:33:15,620 INFO [appenderator_persist_0] io.druid.segment.realtime.appenderator.AppenderatorImpl - Deleting Index File[var/druid/task/index_kafka_aweme_low_frequency_all_event_da2a8411a415b8b_dfjkpjih/work/persist/aweme_low_frequency_all_event_2018-12-08T01:00:00.000Z_2018-12-08T02:00:00.000Z_2018-12-09T16:40:28.781Z_54] 2018-12-09 20:33:15,604 ERROR [task-runner-0-priority-0] io.druid.indexing.kafka.KafkaIndexTask - Encountered exception in run() before persisting. io.druid.java.util.common.ISE: No sink for identifier: aweme_low_frequency_all_event_2018-12-08T01:00:00.000Z_2018-12-08T02:00:00.000Z_2018-12-09T16:40:28.781Z_54 at io.druid.segment.realtime.appenderator.AppenderatorImpl.persistAll(AppenderatorImpl.java:434) ~[druid-server-0.12.2.jar:0.12.2] at io.druid.segment.realtime.appenderator.AppenderatorImpl.push(AppenderatorImpl.java:553) ~[druid-server-0.12.2.jar:0.12.2] at io.druid.segment.realtime.appenderator.BaseAppenderatorDriver.pushInBackground(BaseAppenderatorDriver.java:470) ~[druid-server-0.12.2.jar:0.12.2] at io.druid.segment.realtime.appenderator.StreamAppenderatorDriver.publish(StreamAppenderatorDriver.java:265) ~[druid-server-0.12.2.jar:0.12.2] at io.druid.indexing.kafka.KafkaIndexTask.publishAndRegisterHandoff(KafkaIndexTask.java:1243) ~[druid-kafka-indexing-service-0.12.2.jar:0.12.2] at io.druid.indexing.kafka.KafkaIndexTask.maybePersistAndPublishSequences(KafkaIndexTask.java:1373) ~[druid-kafka-indexing-service-0.12.2.jar:0.12.2] at io.druid.indexing.kafka.KafkaIndexTask.run(KafkaIndexTask.java:544) [druid-kafka-indexing-service-0.12.2.jar:0.12.2] at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:444) [druid-indexing-service-0.12.2.jar:0.12.2] at io.druid.indexing.overlord.ThreadPoolTaskRunner$ThreadPoolTaskRunnerCallable.call(ThreadPoolTaskRunner.java:416) [druid-indexing-service-0.12.2.jar:0.12.2] at java.util.concurrent.FutureTask.run(FutureTask.java:266) [?:1.8.0_91] at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) [?:1.8.0_91] at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) [?:1.8.0_91] at java.lang.Thread.run(Thread.java:745) [?:1.8.0_91] 2018-12-09 20:33:15,622 INFO [task-runner-0-priority-0] io.druid.indexing.kafka.KafkaIndexTask - Persisting all pending data 2018-12-09 20:33:15,623 INFO [task-runner-0-priority-0] io.druid.segment.realtime.appenderator.StreamAppenderatorDriver - Persisting data. 2018-12-09 20:33:15,623 INFO [task-runner-0-priority-0] io.druid.segment.realtime.appenderator.AppenderatorImpl - Submitting persist runnable for dataSource[aweme_low_frequency_all_event] ``` According the log and checked the code, I found why this happened. See below, code 1 is `AppenderatorImpl#persistAll`, code 2 is `AppenderatorImpl#abandonSegment`. Although `sinks` is a ConcurrentHashMap, if `sinks.keySet()` is called at code 1 and after which `sinks.remove(identifier)` is called at code 2, then the for loop go on at code 1, `sinks.get(identifier)` would try to get a object which has been removed. Finally, an exception throwed and task failed. I think we should add an extra lock to solve this bug. - code 1 ```java public ListenableFuture<Object> persistAll(@Nullable final Committer committer) { .... for (SegmentIdentifier identifier : sinks.keySet()) { final Sink sink = sinks.get(identifier); if (sink == null) { throw new ISE("No sink for identifier: %s", identifier); } ..... } ..... } ``` - code 2 ```java private ListenableFuture<?> abandonSegment( final SegmentIdentifier identifier, final Sink sink, final boolean removeOnDiskData ) { ..... log.info("Removing sink for segment[%s].", identifier); sinks.remove(identifier); ..... } ```
---------------------------------------------------------------- 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: [email protected] With regards, Apache Git Services --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
