[fix][io] Not restart instance when kafka source poll exception.

[shibd]

Motivation

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

After #20424, it just closes the Kafka source connector poll thread, and not will trigger restart instance.

Modifications

• When Kafka consumer gets an exception, call notifyError to notify this exception, and when the function framework next calls consumer.read method, it will trigger instance restart.

Verifying this change

• Change closeConnectorWhenUnexpectedExceptionThrownTest to cover it.

Documentation

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

Matching PR in forked repository

[codecov-commenter]

Codecov Report

❌ Patch coverage is 55.55556% with 8 lines in your changes missing coverage. Please review.
✅ Project coverage is 73.12%. Comparing base (f69e7dd) to head (7d8f05c).
⚠️ Report is 1814 commits behind head on master.

Files with missing lines	Patch %	Lines
...rg/apache/pulsar/io/kafka/KafkaAbstractSource.java	55.55%	6 Missing and 2 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             master   #20795       +/-   ##
=============================================
+ Coverage     36.82%   73.12%   +36.30%     
- Complexity    12123    32128    +20005     
=============================================
  Files          1689     1867      +178     
  Lines        129452   139047     +9595     
  Branches      14123    15299     +1176     
=============================================
+ Hits          47675   101685    +54010     
+ Misses        75465    29303    -46162     
- Partials       6312     8059     +1747     

Flag	Coverage Δ
inttests	24.09% <0.00%> (-0.08%)	⬇️
systests	25.02% <0.00%> (+0.02%)	⬆️
unittests	72.41% <55.55%> (+40.38%)	⬆️

Flags with carried forward coverage won't be shown. Click here to find out more.

Files with missing lines	Coverage Δ
...rg/apache/pulsar/io/kafka/KafkaAbstractSource.java	73.28% <55.55%> (+73.28%)	⬆️

... and 1431 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[shibd]

@Technoboy- @poorbarcode Thanks for the feedback.

About this comment: #20795 (comment)

I refactor the start logic:

1. Let consumer.subscribe run on the start thread, which can quickly fail when subscribe exception.
2. Fix InterruptedException will be caught: when encountering an InterruptedException should be thrown it.

PTAL.

[Technoboy-]

LGTM