fix: Return cached pausedPartitionSet

[acktsap]

Resolves: #618

Checklist

• Documentation (if applicable)
• Changelog

[youribonnaffe]

Isn't something we would like to handle in https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ConsumerManager.java the same way metadata are cached between polls?

[jhollandus]

Isn't something we would like to handle in https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/ConsumerManager.java the same way metadata are cached between polls?

This is a good idea. It would be best to split the ToDoubleFunction away from the AtomicInteger and put the integer into the ConsumerManager poll loop. Finally reference the pauses partitions from the consumer manager in the ToDoubleFunction.

[jhollandus]

See #620 (comment)

[what-the-diff]

PR Summary

• Fixes added to CHANGELOG.adoc
  Two bug fixes have been implemented. They are primarily aimed at addressing the issue of the parallel consumer sometimes stopping data processing, and also facilitating the return of a cached paused partition set.

• Update to BrokerPollSystem.java
  A new feature has been introduced in the ConsumerManager class. Specifically, a method named getPausedPartitionSize has been added. This addition should help to efficiently manage and comprehend the size of paused partitions.

• Modifications in ConsumerManager.java
  Multiple changes have been made to the ConsumerManager class. A field named pausedPartitionSizeCache was added to boost the performance, given it provides caching capabilities. Additionally, the method updateMetadataCache was renamed to updateCache, aligning better with its functionality of routinely updating the cache. Also, a new method getPausedPartitionSize was added to handle the retrieval of paused partition sizes.

• Change to PCModuleTestEnv.java
  To stay consistent with the main code, the method updateMetadataCache in this file has been renamed to updateCache. This reflects the broader scope of work this method undertakes beyond just updating the metadata.

[acktsap]

Hi, thanks for comments. I moved cache logic into ConsumerManager.

[cla-assistant]

All committers have signed the CLA.

[eddyv]

It looks like the latest changes failed a few tests.

I offered some suggestions so that we just keep track of the size itself, which seems to address this issue.

Will need to add the following in BrokerPollSystem.initMetrics() if you choose to go that route.

-                this.consumerManager, consumerManager -> consumerManager.paused().size());
+                this.consumerManager, consumerManager -> consumerManager.getPausedPartitionSize().get());

[eddyv]

lgtm thanks 👍

[jhollandus]

LGTM