[SPARK-29993][SS]Avoid continuous warn log in get record and offset from Kafka

[wenxuanguan]

What changes were proposed in this pull request?

Log warn once when create Kafka consumer to avoid continuous warn log in ss application.

Why are the changes needed?

In streaming application, Kafka consumer run in UninterruptibleThread, in which log warn info every
time consumer get record and offset, making continuous warn log.
Since warn info is for Kafka consumer, and there is no need to log every time when consumer get record and offset, we can log once when create consumer.

Does this PR introduce any user-facing change?

No

How was this patch tested?

N/A

[AmplabJenkins]

Can one of the admins verify this patch?

[HeartSaVioR]

It looks less intuitive for me. I'm not sure how much this makes us be bothered (the message is quite serious one so I'd be even OK if it's a bit bothering), but if it's really bugging, then I'd add an explicit flag to check it.

[wenxuanguan]

@HeartSaVioR In micro-batch application, every mini-batch job will generate two logs per task in executor log file. Since micro-batch application is long running, there will be massive warning log.

[HeartSaVioR]

The thing is, the thread should have been UninterruptibleThread; if we found the other case we should file an issue and try to fix.

[gaborgsomogyi]

I think the issue is more than just logging something here and there. KAFKA-1894 is a serious issue which was fixed in Kafka 2.0.0. Since the change only effects the consumer and Spark has already this version maybe Spark doesn't need to run consumers on UninterruptibleThread.

• Micro-batch processing has still this feature which can be removed after it has been double checked that the mentioned Kafka jira solved all the APIs which are used by Spark.
• Continuous processing I guess not considered KAFKA-1894

So my point is first we must understand on what APIs KAFKA-1894 provided solution and then I see 2 options:

• All the APIs which used by Spark fixed => UninterruptibleThread can be removed.
• NOT all the APIs which used by Spark fixed => Continuous processing must use it

[HeartSaVioR]

KAFKA-1894 will not be resolved with just a version upgrade, as it was closed with KIP-266 as a solution. It requires to change the calls of Kafka API as non-blocking version (Having Duration as parameter), which we still don't migrate. We still call poll() with long parameter, and also wait KIP-396 for replacement of poll(0).

When we migrate we should take the side effect into account that the semantic of timeout is changing - it will include the duration of fetching metadata as well. End users need to set bigger value of timeout, or we should add artificial value to given timeout.

[gaborgsomogyi]

KAFKA-1894 will not be resolved with just a version upgrade

+1

It requires to change the calls of Kafka API as non-blocking version

That's what we can't do all the places at the moment since API is still missing which will be only available in Kafka 2.5.

[wenxuanguan]

@HeartSaVioR @gaborgsomogyi Since warn log is important and need to be logged every time, I will close this PR.
Thanks for your reply.