[SPARK-22606][Streaming]Add threadId to the CachedKafkaConsumer key

[eatoncys]

What changes were proposed in this pull request?

If the value of param 'spark.streaming.concurrentJobs' is more than one, and the value of param 'spark.executor.cores' is more than one, there may be two or more tasks in one executor will use the same kafka consumer at the same time, then it will throw an exception: "KafkaConsumer is not safe for multi-threaded access";
for example:
spark.streaming.concurrentJobs=2
spark.executor.cores=2
spark.cores.max=2
if there is only one topic with one partition('topic1',0) to consume, there will be two jobs to run at the same time, and they will use the same cacheKey('groupid','topic1',0) to get the CachedKafkaConsumer from the cache list of' private var cache: ju.LinkedHashMap[CacheKey, CachedKafkaConsumer[_, _]]' , then it will get the same CachedKafkaConsumer.

this PR add threadId to the CachedKafkaConsumer key to prevent two thread using a consumer at the same time.

How was this patch tested?

existing ut test

[SparkQA]

Test build #84183 has finished for PR 19819 at commit aa02d89.

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

[lvdongr]

Will the cached consumer to the same partition increase , when different tasks consume the same partition and no place to remove?

[gaborgsomogyi]

It will create a new consumer for each thread. This could be quite resource consuming when several topics shared with thread pools.

[gaborgsomogyi]

@lvdongr I think this can be closed as the problem solved.

[lvdongr]

I've seen your PR: #20997, a good solution @gaborgsomogyi