[SPARK-33635][SS] Adjust the order of check in KafkaTokenUtil.needTokenUpdate to remedy perf regression

[HeartSaVioR]

What changes were proposed in this pull request?

This PR proposes to adjust the order of check in KafkaTokenUtil.needTokenUpdate, so that short-circuit applies on the non-delegation token cases (insecure + secured without delegation token) and remedies the performance regression heavily.

Why are the changes needed?

There's a serious performance regression between Spark 2.4 vs Spark 3.0 on read path against Kafka data source.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Manually ran a reproducer (https://github.com/codegorillauk/spark-kafka-read with modification to just count instead of writing to Kafka topic) with measuring the time.

the branch applying the change with adding measurement

https://github.com/HeartSaVioR/spark/commits/debug-SPARK-33635-v3.0.1

the branch only adding measurement

https://github.com/HeartSaVioR/spark/commits/debug-original-ver-SPARK-33635-v3.0.1

the result (before the fix)

count: 10280000
Took 41.634007047 secs

21/01/06 13:16:07 INFO KafkaDataConsumer: debug ver. 17-original
21/01/06 13:16:07 INFO KafkaDataConsumer: Total time taken to retrieve: 82118 ms

the result (after the fix)

count: 10280000
Took 7.964058475 secs

21/01/06 13:08:22 INFO KafkaDataConsumer: debug ver. 17
21/01/06 13:08:22 INFO KafkaDataConsumer: Total time taken to retrieve: 987 ms

[HeartSaVioR]

I just made the smallest change as a short-term solution (the problem will persist for end users using Kafka delegation token), as Spark 3.1.0 RC vote is happening and I don't want to drag the release too much.

It would be ideal if we deal with long-term solution (making overhead of HadoopDelegationTokenManager.isServiceEnabled be minor, or reduce occurrence of the check heavily) in time, but just wanted to be safe.

[HeartSaVioR]

The further optimization should be depending on the possibility of "changes" in SparkConf on the fly. If we don't allow it (or at least Kafka/token related configuration), we just need to call HadoopDelegationTokenManager.isServiceEnabled once per JVM, and cache it to avoid calling per each get.

[HeartSaVioR]

cc. @tdas @zsxwing @jose-torres @viirya @gaborgsomogyi @xuanyuanking

[HeartSaVioR]

In addition, the time in Total time taken to retrieve checks pure overhead compared to Spark 2.4, but I ran the application with my local dev where parallelism is not fully enabled (just ran with local[3]) and I also have to do some other thing in parallel, so the value should be pretty much smaller (probably trivial) if the test environment is isolated properly with enough power to enable full parallelism.

[viirya]

As a quick fix, it looks good as it just reorders the conditions. Even we will deal with long-term solution, I think it is still no harm to have the quick fix now.

[SparkQA]

Test build #133711 has finished for PR 31056 at commit f5decdd.

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

[SparkQA]

Kubernetes integration test starting
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38299/

[dongjoon-hyun]

+1, LGTM. Thank you, @HeartSaVioR and @viirya . I agree with you with this fix.
Merged to master/3.1/3.0.

[dongjoon-hyun]

cc @HyukjinKwon since he is the release manager of Spark 3.1.0.

[SparkQA]

Kubernetes integration test status success
URL: https://amplab.cs.berkeley.edu/jenkins/job/SparkPullRequestBuilder-K8s/38299/

[gaborgsomogyi]

Late LGTM.

[gaborgsomogyi]

Since SparkConf is static we can cache the first time calculated values. I've started to work on the long term solution.