[SPARK-32033][SS][DSTEAMS]Use new poll API in Kafka connector executor side to avoid infinite wait

[gaborgsomogyi]

What changes were proposed in this pull request?

Spark uses an old and deprecated API named KafkaConsumer.poll(long) which never returns and stays in live lock if metadata is not updated (for instance when broker disappears at consumer creation). Please see Kafka documentation and standalone test application for further details.

In this PR I've applied the new KafkaConsumer.poll(Duration) API on executor side. Please note driver side still uses the old API which will be fixed in SPARK-32032.

Why are the changes needed?

Infinite wait in KafkaConsumer.poll(long).

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing unit tests.

[gaborgsomogyi]

cc @zsxwing @HeartSaVioR

[SparkQA]

Test build #124278 has finished for PR 28871 at commit d33a5f6.

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

[HeartSaVioR]

The code change looks good.

Since we're here and this changes the semantic of timeout, could we double-check the default value of timeout is enough, and the documentation is correct?

I roughly skimmed it and found the option follows spark.network.timeout as the default value which is 120s by default, but the default value of kafkaConsumer.pollTimeoutMs in integration doc is 512 ms. Would be nice to fix either one, and probably we would want to higher one, as now it counts metadata update.

I guess it'd be nice to also update the doc as well as migration guide (not 100% sure about this, so let's hear other's voice) so that the semantic of timeout now includes metadata update, so end users will want to set higher value than before.

[gaborgsomogyi]

The reason why not added any increase in the kafkaConsumer.pollTimeoutMs area because I remembered always that it falls back to spark.network.timeout which is high enough.

Honestly didn't check the doc, presumed it's set properly. Nice catch! Fixed it. :)

[gaborgsomogyi]

Related the migration guide I don't think it's required. What would you write there, which represents real value? Presumable you've set kafkaConsumer.pollTimeoutMs high enough but if it fails w/ timeout increase it more? :)

I've seen setups which contain enough room to handle such situations + admins know this issue well enough not to write this down.

[SparkQA]

Test build #124288 has finished for PR 28871 at commit 42e4adb.

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

[dongjoon-hyun]

+1, LGTM. Thank you, @gaborgsomogyi and @HeartSaVioR .
Merged to master.

[gaborgsomogyi]

Thank you @dongjoon-hyun and @HeartSaVioR for the quick action.
We've solved a long standing cluster stop issue.