[improve][io] Add notifyError method on PushSource.

[shibd]

Motivation

If one source connector extends PushSource. An exception was encountered when this source connector consumed the source system data, it had no way to notify the function framework to restart this connector.

Modifications

• Abstract BatchPushSource logic to AbstractPushSource.
• Let PushSource to extends AbstractPushSource to extend a new method(notifyError).

Verifying this change

• Add PushSourceTest to cover it.

Documentation

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

Matching PR in forked repository

[codecov-commenter]

Codecov Report

Merging #20791 (a8af71f) into master (e96b339) will increase coverage by 39.65%.
The diff coverage is 80.64%.

@@              Coverage Diff              @@
##             master   #20791       +/-   ##
=============================================
+ Coverage     33.44%   73.09%   +39.65%     
- Complexity    12097    32115    +20018     
=============================================
  Files          1612     1867      +255     
  Lines        126568   139040    +12472     
  Branches      13824    15294     +1470     
=============================================
+ Hits          42328   101629    +59301     
+ Misses        78686    29346    -49340     
- Partials       5554     8065     +2511     

Flag	Coverage Δ
inttests	24.07% <0.00%> (-0.02%)	⬇️
systests	25.05% <3.22%> (?)
unittests	72.37% <80.64%> (+40.43%)	⬆️

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

Impacted Files	Coverage Δ
...rg/apache/pulsar/broker/service/BrokerService.java	81.82% <ø> (+26.02%)	⬆️
.../org/apache/pulsar/io/core/AbstractPushSource.java	76.92% <76.92%> (ø)
...rg/apache/pulsar/client/impl/TopicListWatcher.java	66.66% <100.00%> (+43.47%)	⬆️
...ava/org/apache/pulsar/io/core/BatchPushSource.java	100.00% <100.00%> (+100.00%)	⬆️
...ain/java/org/apache/pulsar/io/core/PushSource.java	100.00% <100.00%> (+100.00%)	⬆️

... and 1515 files with indirect coverage changes

[RobertIndie]

@shibd Would this PR change the interface for the ecosystem component? If yes, I think we shouldn't add it to branch-3.0.
I remove the label release-3.0.2 first. But feel free to confirm it and add back the label if it's suitable to cherry-pick to the LTS branch.

[shibd]

@shibd Would this PR change the interface for the ecosystem component? If yes, I think we shouldn't add it to branch-3.0. I remove the label release-3.0.2 first. But feel free to confirm it and add back the label if it's suitable to cherry-pick to the LTS branch.

Thanks for your reminder. I remove the release-x label, and patch a PIP design: #20807

[nlu90]

We probably also need this in Sink connector to capture error happens when the actual write action happens

[shibd]

We probably also need this in Sink connector to capture error happens when the actual write action happens

Sure, I'll research a way to do it.