Fixes #3306.

Motivation

Kafka connectors may raise two exceptions during the polling process.

1. when the consumer polls but is not subscribed to any topic, the thread raises an exception and the consumer shuts down permanently.
2. RecordDeserializationException, which happens when the consumer fails to deserialize a record.

Modifications

For the first problem, I added a check on the subscription of the consumer.
The second issue is a bit more tricky. It was a bug in initializing the Kafka consumer. The added lines will fix the bug for the following reasons:

kafka client use Class.forName(trimmed, true, Utils.getContextOrKafkaClassLoader()) to get the Class object, and the create the instance, the key point is the classLoader, which is specified by the last param, the implementation of method Utils.getContextOrKafkaClassLoader() is

public static ClassLoader getContextOrKafkaClassLoader() {
    ClassLoader cl = Thread.currentThread().getContextClassLoader();
    if (cl == null)
        return getKafkaClassLoader();
    else
        return cl;
}

Documentation

• Does this pull request introduce a new feature?
• No
• If yes, how is the feature documented? N/A
• If a feature is not applicable for documentation, explain why?
• If a feature is not documented yet in this PR, please create a followup issue for adding the documentation