[SPARK-15272] DirectKafkaInputDStream doesn't work with window operation

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Incomplete
• Affects Version/s: 1.5.2
• Component/s: DStreams
• Labels: bulk-closed
• Assignee: Unassigned
• Reporter: Lubomir Nerad
• Created: 11/May/16 16:19
• Updated: 21/May/19 04:32
• Resolved: 21/May/19 04:32

Description

Using Kafka direct DStream with simple window operation like:

kafkaDStream.window(Durations.milliseconds(10000),
                    Durations.milliseconds(1000));
            .print();

with 1s batch duration either freezes after several seconds or lags terribly (depending on cluster mode).

This happens when Kafka brokers are not part of the Spark cluster (they are on different nodes). The KafkaRDD still reports them as preferred locations. This doesn't seem to be problem in non-window scenarios but with window it conflicts with delay scheduling algorithm implemented in TaskSetManager. It either significantly delays (Yarn mode) or completely drains (Spark mode) resource offers with TaskLocality.ANY which are needed to process tasks with these Kafka broker aligned preferred locations. When delay scheduling algorithm is switched off (spark.locality.wait=0), the example works correctly.

I think that the KafkaRDD shouldn't report preferred locations if the brokers don't correspond to worker nodes or allow the reporting of preferred locations to be switched off. Also it would be good if delay scheduling algorithm didn't drain / delay offers in the case, the tasks have unmatched preferred locations.