[SPARK-24713]AppMatser of spark streaming kafka OOM if there are hund…

[yuanboliu]

We have hundreds of kafka topics need to be consumed in one application. The application master will throw OOM exception after hanging for nearly half of an hour.

OOM happens in the env with a lot of topics, and it's not convenient to set up such kind of env in the unit test. So I didn't change/add test case.

[yuanboliu]

Can one of the admins verify this patch?

[yuanboliu]

@koeninger Sorry to interrupt, would you please review my patch? Thanks in advance.

[koeninger]

Jenkins, ok to test

[koeninger]

@yuanboliu Can you clarify why repeated pause is necessary?

[SparkQA]

Test build #92528 has finished for PR 21690 at commit 15d03e6.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[yuanboliu]

@koeninger Thanks for your review

The first pause is used to stop poll() in the method paranoidPoll
The second one is attached because of p.partition().
I'm not sure whether the state of pause will be rewritten after these methods are called, so I use pause repeatedly.

[koeninger]

@yuanboliu From reading KafkaConsumer code, and from testing, I don't see where consumer.position() alone would un-pause topicpartitions. See below. Can you give a counter-example?

I am seeing poll() reset the paused state. When you are having the problem, are you seeing the info level log messages "poll(0) returned messages"?

If that's what's happening, I think the best we can do is call pause() in only one place, the first line of paranoidPoll, e.g.

c.pause(c.assignment)

val msgs = c.poll(0)

Here's what I saw in testing:

scala> c.paused
res34: java.util.Set[org.apache.kafka.common.TopicPartition] = []

scala> c.assignment
res35: java.util.Set[org.apache.kafka.common.TopicPartition] = [test-0]

scala> c.pause(topics)

scala> c.paused
res37: java.util.Set[org.apache.kafka.common.TopicPartition] = [test-0]

scala> c.position(tp)
res38: Long = 248

scala> c.paused
res39: java.util.Set[org.apache.kafka.common.TopicPartition] = [test-0]

scala> c.poll(0)
res40: org.apache.kafka.clients.consumer.ConsumerRecords[String,String] = org.apache.kafka.clients.consumer.ConsumerRecords@20d7efbe

scala> c.paused
res41: java.util.Set[org.apache.kafka.common.TopicPartition] = [test-0]

scala> c.position(tp)
res42: Long = 248

scala> c.paused
res43: java.util.Set[org.apache.kafka.common.TopicPartition] = [test-0]

scala> c.poll(1)
res44: org.apache.kafka.clients.consumer.ConsumerRecords[String,String] = org.apache.kafka.clients.consumer.ConsumerRecords@20d7efbe

scala> c.paused
res45: java.util.Set[org.apache.kafka.common.TopicPartition] = [test-0]

scala> c.poll(100)
res46: org.apache.kafka.clients.consumer.ConsumerRecords[String,String] = org.apache.kafka.clients.consumer.ConsumerRecords@28e4439b

scala> c.paused
res47: java.util.Set[org.apache.kafka.common.TopicPartition] = []

[yuanboliu]

@koeninger Thanks for your details. Sorry quite busy this week. I will delete the last pause, test the patch on my own cluster this weekend and give feedback asap.

[koeninger]

@yuanboliu What I'm suggesting is more like this:

https://github.com/apache/spark/compare/master...koeninger:SPARK-24713?expand=1

[yuanboliu]

@koeninger Thanks for your reply. Agree with you. there is no need to to use pause repeatedly.
This is my test without any pause, and the app master stuck for a long time without any process

I will update my patch shortly.

[SparkQA]

Test build #92709 has finished for PR 21690 at commit d1a8c60.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[yuanboliu]

@koeninger Sorry to interrupt, could you take a look at my patch?

[koeninger]

What results are you seeing?

…

On Thu, Jul 12, 2018, 6:53 AM Yuanbo Liu ***@***.***> wrote: @koeninger <https://github.com/koeninger> Sorry to interrupt, could you take a look at my patch? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <#21690 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AAGABxNyVlFhIo7z2D7jXtfeBfsVRsMmks5uF0bXgaJpZM4U-p4M> .

[yuanboliu]

After applying this patch, my application can be running successfully. This issue could happen in the case of many topics(hundreds of ) consumed.

[koeninger]

LGTM, merging to master. Thanks!

[yuanboliu]

Thanks very much