Kafka support manual partition pause/resume with pause-if-no-requests=true

[ozangunalp]

Fixes #3295

[codecov]

Codecov Report

❌ Patch coverage is 81.81818% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 77.93%. Comparing base (a96442f) to head (96a4098).
⚠️ Report is 1257 commits behind head on main.

Files with missing lines	Patch %	Lines
...ve/messaging/kafka/impl/ReactiveKafkaConsumer.java	81.25%	1 Missing and 5 partials ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main    #3538      +/-   ##
============================================
+ Coverage     77.47%   77.93%   +0.46%     
- Complexity     3778     5641    +1863     
============================================
  Files           306      487     +181     
  Lines         12673    18842    +6169     
  Branches       1648     2319     +671     
============================================
+ Hits           9818    14685    +4867     
- Misses         2116     3009     +893     
- Partials        739     1148     +409     

Files with missing lines	Coverage Δ
...ye/reactive/messaging/kafka/i18n/KafkaLogging.java	100.00% <ø> (ø)
...ctive/messaging/kafka/impl/RebalanceListeners.java	92.30% <100.00%> (+6.34%)	⬆️
...ve/messaging/kafka/impl/ReactiveKafkaConsumer.java	80.33% <81.25%> (-1.61%)	⬇️

... and 292 files with indirect coverage changes

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[cescoffier]

Is this correct? removeAll returns true if it removes at least one partition from the collection. I feel like you want to remove all of them. If partitions {tp0, tp1, tp2} are revoked but only tp0 was manually paused, the log says "Cleaning up paused partitions [tp0, tp1, tp2] due to revocation". Slightly misleading, no?.

[ozangunalp]

Nice catch, we should log the intersection if there are any.

[ozangunalp]

Instead I added both revoked and paused partitions to the log.

[cescoffier]

I don't believe it's correct.
removeAll mutates manuallyPausedPartitions before the log call. The log message template is: "Cleaning up manually paused partitions %s due to revocation of %s".

The first %s receives the remaining set, not the cleaned-up set.

I think you need:

void removeManuallyPausedPartitions(Collection<TopicPartition> partitions) {
    Set<TopicPartition> removed = new HashSet<>(manuallyPausedPartitions);
    removed.retainAll(partitions);
    if (!removed.isEmpty()) {
        manuallyPausedPartitions.removeAll(removed);
        log.cleaningUpPausedPartitions(removed, partitions);
    }
}

[cescoffier]

or update the log messsage: "Remaining manually paused partitions %s after revocation of %s"

[ozangunalp]

Yes, let's change the log message.