fix : for issue 606 : exclude stale containers from counting and add remove add remove staled work containers function in ProcessingShard

[Martyn Ye (sangreal)]

Description...
This is the fix for #606

The issue happens when rebalancing and partitions got re-assigned:

1. The workcontainer will be marked as stale container since its epoch is mismatch between partitionsAssignmentEpoch of PartitionState and epoch from WorkContainer
   https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/PartitionState.java#L655

2. inFlight of WorkContainer will be stay false since they will be filtered to be selected in runUserFunction in AbstractParallelEoSStreamProcessor
   https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/AbstractParallelEoSStreamProcessor.java#L1256

3. The consumer paused will be triggered since if the number of staled messages are big enough isSufficientlyLoaded in WorkManager will return true.
   https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkManager.java#L232C1-L232C1
   https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/WorkContainer.java#L230

4. Since the staled workcontainer can never be selected to run user function. The consumer will be paused infinitely until next-time rebalancing
   https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/internal/BrokerPollSystem.java#L304

5. Meanwhile, while getWorkIfAvailable in WorkManager will try to extract active workcontainers, but since the staled containers have the lower offsets, they will be extracted always earlier than the new ones. (if the same partitions have been assigned to the consumer again). Then the new records have no chances to be processed since the it will break out of the while loop.
   https://github.com/confluentinc/parallel-consumer/blob/master/parallel-consumer-core/src/main/java/io/confluent/parallelconsumer/state/ProcessingShard.java#L134

The proposed solution:

1. We should exclude those stale work containers from calculation
2. remove staled work containers from shard when partition assigned and revoke happen

Checklist

• Documentation (if applicable)
• Changelog

[what-the-diff]

PR Summary

• Updated Documentation
  The documentation files CHANGELOG.adoc and README.adoc have been updated to include information regarding the new version 0.5.2.7 and a fix for an issue with parallel consumer processing (Parallel consumer stops processing data sometimes #606).

• Improvements to Partition State Management
  In PartitionStateManager.java, a clean-up operation has been included to remove any old or "stale" partition data. This operation will be performed when partitions are assigned and also when they are removed.

• New Workflow Control in Processing Shard
  A new function removeStaleWorkContainersFromShard was added in ProcessingShard.java. It's designed to clear out any outdated work containers from the processing shard, ensuring that we only work with the most recent data. This function will be employed during the allocation of new work.

• Enhanced Shard Management
  A new function removeStaleContainers was added to ShardManager.java. This function clears out any outdated (stale) containers and records when these containers are removed. This ensures our system is always dealing with fresh and relevant data.

[Edward Vaisman (eddyv)]

LGTM

[Martyn Ye (sangreal)]

Edward Vaisman (@eddyv) Thanks a lot for your quick response. Actually we are kind of blocked by this issue, when could we expect this fix merged?
And when could we have a snapshot version or release version to use?

[Edward Vaisman (eddyv)]

Hey! Allow us some time to do a quick cleanup of some open issues / pr's, and we'll get a new version out asap. Thanks for all the work and effort put into this.

[Martyn Ye (sangreal)]

Edward Vaisman (@eddyv) Thank you so much for your help here! We are rolling out parallel consumer to company wide usage, this fix is very crucial for us.
Thanks again and looking forward to more interaction in the future.