Close Pulsar consumer correctly when fetching offsets

[KKcorps]

Instructions:

1. The PR has to be tagged with at least one of the following labels (*):
   1. feature
   2. bugfix
   3. performance
   4. ui
   5. backward-incompat
   6. release-notes (**)
2. Remove these instructions before publishing the PR.

(*) Other labels to consider:

• testing
• dependencies
• docker
• kubernetes
• observability
• security
• code-style
• extension-point
• refactor
• cleanup

(**) Use release-notes label for scenarios like:

• New configuration options
• Deprecation of configurations
• Signature changes to public methods/interfaces
• New plugins added or old plugins removed

[codecov-commenter]

Codecov Report

Merging #8803 (cb41167) into master (73316a6) will increase coverage by 1.17%. The diff coverage is n/a.

@@             Coverage Diff              @@
##             master    #8803      +/-   ##
============================================
+ Coverage     68.51%   69.69%   +1.17%     
- Complexity     4601     4618      +17     
============================================
  Files          1736     1735       -1     
  Lines         91196    91253      +57     
  Branches      13633    13635       +2     
============================================
+ Hits          62480    63595    +1115     
+ Misses        24330    23236    -1094     
- Partials       4386     4422      +36     

Flag	Coverage Δ
integration1	26.95% <ø> (-0.09%)	:arrow_down:
integration2	25.76% <ø> (?)
unittests1	66.19% <ø> (+0.04%)	:arrow_up:
unittests2	14.18% <ø> (+0.01%)	:arrow_up:

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

Impacted Files	Coverage Δ
...data/manager/realtime/DefaultSegmentCommitter.java	0.00% <0.00%> (-80.00%)	:arrow_down:
...er/api/resources/LLCSegmentCompletionHandlers.java	43.56% <0.00%> (-18.82%)	:arrow_down:
...mmon/request/context/predicate/NotEqPredicate.java	76.92% <0.00%> (-7.70%)	:arrow_down:
...altime/ServerSegmentCompletionProtocolHandler.java	51.88% <0.00%> (-6.61%)	:arrow_down:
...not/broker/broker/helix/ClusterChangeMediator.java	75.26% <0.00%> (-5.38%)	:arrow_down:
...nction/DistinctCountBitmapAggregationFunction.java	41.96% <0.00%> (-5.19%)	:arrow_down:
...ion/function/DistinctCountAggregationFunction.java	48.43% <0.00%> (-4.94%)	:arrow_down:
.../org/apache/pinot/core/startree/StarTreeUtils.java	76.28% <0.00%> (-2.07%)	:arrow_down:
.../predicate/NotEqualsPredicateEvaluatorFactory.java	71.15% <0.00%> (-1.93%)	:arrow_down:
...tream/kafka20/KafkaPartitionLevelStreamConfig.java	63.15% <0.00%> (-1.85%)	:arrow_down:
... and 150 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more Δ = absolute <relative> (impact), ø = not affected, ? = missing data Powered by Codecov. Last update 73316a6...cb41167. Read the comment docs.

[Jackie-Jiang]

Fixed with #9859