[fix][io][branch-2.11] Not restart instance when kafka source poll ex…

[shibd]

Motivation

When Kafka source poll message exception, it will throw it to the function framework, and not trigger the restart function instance.

#20795 uses a new method on PulsarSource #20791 to fix it on master branch. To maintain compatibility, we can't cherry-pick it to the historical version, so the class is implemented in the Kafka source connector.

Modifications

• Let consumer.subscribe run on the start thread, which can quickly fail when subscribe exception.
• Add new class KafkaPushSource on Kafka source connector.
• When Kafka consumer exception, it will call notifyError to report the exception to the io connector framework.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository