fix(consumer): maintain ordering of offset commit requests #2947

prestona · 2024-07-23T17:51:10Z

Use the Broker.lock to prevent concurrent calls to commit from potentially being re-ordered between determining which offsets to include in the offset commit request, and sending the request to Kafka. This has the benefit that the lock is only held while the OffsetCommitRequest is sent, not for the period of time that Sarama is waiting for the response from the broker, improving the rate at which commit requests can be made.

The down side is that (without further re-factoring to the Broker) this change duplicates some of the code from the Broker into the offset manager. Potentially, making the code more fragile.

Hold broker lock to prevent concurrent calls to commit from potentially being re-ordered between determining which offsets to include in the offset commit request, and sending the request to Kafka. Move finding the coordinator before creating the request to avoid holding the lock in the case the coordinator is unknown. This requires a change to the "new offset manager" test, which previously didn't expect the mock broker to receive a find coordinator request. Fixes: IBM#2940 Co-authored-by: Michael Burgess <michburg@uk.ibm.com> Signed-off-by: Adrian Preston <PRESTONA@uk.ibm.com>

This test simulates multiple concurrent commits for a single group running in a single process. The expected behavior is that commits are delivered in order to the mock broker, with no offset in the OffsetCommitRequest being for a lower value than previously committed for the partition. Signed-off-by: Adrian Preston <PRESTONA@uk.ibm.com>

offset_manager.go

Signed-off-by: Adrian Preston <PRESTONA@uk.ibm.com>

Specifically: - Simplify variable declarations for zero-value values - Use sync/atomic.Pointer instead of sync.Mutex - t.Error -> t.Fatal to avoid continuing and de-referencing nil - Remove unused named return values (copy and paste error) Signed-off-by: Adrian Preston <PRESTONA@uk.ibm.com>

offset_manager_test.go

dnwe

LGTM. After some further review and discussion lets go ahead and merge this version of the fix

[![Mend Renovate](https://app.renovatebot.com/images/banner.svg)](https://renovatebot.com) This PR contains the following updates: | Package | Change | Age | Adoption | Passing | Confidence | |---|---|---|---|---|---| | [github.com/IBM/sarama](https://github.com/IBM/sarama) | `v1.43.2` -> `v1.43.3` | [![age](https://developer.mend.io/api/mc/badges/age/go/github.com%2fIBM%2fsarama/v1.43.3?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![adoption](https://developer.mend.io/api/mc/badges/adoption/go/github.com%2fIBM%2fsarama/v1.43.3?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![passing](https://developer.mend.io/api/mc/badges/compatibility/go/github.com%2fIBM%2fsarama/v1.43.2/v1.43.3?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![confidence](https://developer.mend.io/api/mc/badges/confidence/go/github.com%2fIBM%2fsarama/v1.43.2/v1.43.3?slim=true)](https://docs.renovatebot.com/merge-confidence/) | --- > [!WARNING] > Some dependencies could not be looked up. Check the Dependency Dashboard for more information. --- ### Release Notes <details> <summary>IBM/sarama (github.com/IBM/sarama)</summary> ### [`v1.43.3`](https://github.com/IBM/sarama/releases/tag/v1.43.3): Version 1.43.3 (2024-08-12) [Compare Source](https://github.com/IBM/sarama/compare/v1.43.2...v1.43.3)  #### What's Changed ##### 🐛 Fixes - fix: declare assignor variable for examples & clean up log format by [@kumakichi](https://github.com/kumakichi) in [https://github.com/IBM/sarama/pull/2909](https://github.com/IBM/sarama/pull/2909) - fix(consumer): maintain ordering of offset commit requests by [@prestona](https://github.com/prestona) in [https://github.com/IBM/sarama/pull/2947](https://github.com/IBM/sarama/pull/2947) - fix(producer): treat ErrKafkaStorageError as retriable by [@richardartoul](https://github.com/richardartoul) in [https://github.com/IBM/sarama/pull/2939](https://github.com/IBM/sarama/pull/2939) ##### 📦 Dependency updates - chore(deps): bump the golang-org-x group across 1 directory with 2 updates by [@dependabot](https://github.com/dependabot) in [https://github.com/IBM/sarama/pull/2956](https://github.com/IBM/sarama/pull/2956) - chore(deps): bump github.com/eapache/go-resiliency from 1.6.0 to 1.7.0 by [@dependabot](https://github.com/dependabot) in [https://github.com/IBM/sarama/pull/2944](https://github.com/IBM/sarama/pull/2944) - chore(deps): bump github.com/klauspost/compress from 1.17.8 to 1.17.9 by [@dependabot](https://github.com/dependabot) in [https://github.com/IBM/sarama/pull/2926](https://github.com/IBM/sarama/pull/2926) ##### 🔧 Maintenance - fix(ci): correct docker-compose install by [@dnwe](https://github.com/dnwe) in [https://github.com/IBM/sarama/pull/2954](https://github.com/IBM/sarama/pull/2954) ##### 📝 Documentation - fix(doc): correct JVM's config name corresponding to MaxWaitTime by [@abhipranay](https://github.com/abhipranay) in [https://github.com/IBM/sarama/pull/2893](https://github.com/IBM/sarama/pull/2893) #### New Contributors - [@abhipranay](https://github.com/abhipranay) made their first contribution in [https://github.com/IBM/sarama/pull/2893](https://github.com/IBM/sarama/pull/2893) - [@kumakichi](https://github.com/kumakichi) made their first contribution in [https://github.com/IBM/sarama/pull/2909](https://github.com/IBM/sarama/pull/2909) - [@richardartoul](https://github.com/richardartoul) made their first contribution in [https://github.com/IBM/sarama/pull/2939](https://github.com/IBM/sarama/pull/2939) **Full Changelog**: IBM/sarama@v1.43.2...v1.43.3 </details> --- ### Configuration 📅 **Schedule**: Branch creation - "on tuesday" (UTC), Automerge - At any time (no schedule defined). 🚦 **Automerge**: Disabled by config. Please merge this manually once you are satisfied. ♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox. 🔕 **Ignore**: Close this PR and you won't be reminded about this update again. --- - [ ] If you want to rebase/retry this PR, check this box --- This PR was generated by [Mend Renovate](https://www.mend.io/free-developer-tools/renovate/). View the [repository job log](https://developer.mend.io/github/open-telemetry/opentelemetry-collector-contrib).  --------- Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> Co-authored-by: opentelemetrybot <107717825+opentelemetrybot@users.noreply.github.com> Co-authored-by: Yang Song <songy23@users.noreply.github.com>

[![Mend Renovate](https://app.renovatebot.com/images/banner.svg)](https://renovatebot.com) This PR contains the following updates: | Package | Change | Age | Adoption | Passing | Confidence | |---|---|---|---|---|---| | [github.com/IBM/sarama](https://github.com/IBM/sarama) | `v1.43.2` -> `v1.43.3` | [![age](https://developer.mend.io/api/mc/badges/age/go/github.com%2fIBM%2fsarama/v1.43.3?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![adoption](https://developer.mend.io/api/mc/badges/adoption/go/github.com%2fIBM%2fsarama/v1.43.3?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![passing](https://developer.mend.io/api/mc/badges/compatibility/go/github.com%2fIBM%2fsarama/v1.43.2/v1.43.3?slim=true)](https://docs.renovatebot.com/merge-confidence/) | [![confidence](https://developer.mend.io/api/mc/badges/confidence/go/github.com%2fIBM%2fsarama/v1.43.2/v1.43.3?slim=true)](https://docs.renovatebot.com/merge-confidence/) | --- > [!WARNING] > Some dependencies could not be looked up. Check the Dependency Dashboard for more information. --- ### Release Notes <details> <summary>IBM/sarama (github.com/IBM/sarama)</summary> ### [`v1.43.3`](https://github.com/IBM/sarama/releases/tag/v1.43.3): Version 1.43.3 (2024-08-12) [Compare Source](https://github.com/IBM/sarama/compare/v1.43.2...v1.43.3)  #### What's Changed ##### 🐛 Fixes - fix: declare assignor variable for examples & clean up log format by [@&open-telemetry#8203;kumakichi](https://github.com/kumakichi) in [https://github.com/IBM/sarama/pull/2909](https://github.com/IBM/sarama/pull/2909) - fix(consumer): maintain ordering of offset commit requests by [@&open-telemetry#8203;prestona](https://github.com/prestona) in [https://github.com/IBM/sarama/pull/2947](https://github.com/IBM/sarama/pull/2947) - fix(producer): treat ErrKafkaStorageError as retriable by [@&open-telemetry#8203;richardartoul](https://github.com/richardartoul) in [https://github.com/IBM/sarama/pull/2939](https://github.com/IBM/sarama/pull/2939) ##### 📦 Dependency updates - chore(deps): bump the golang-org-x group across 1 directory with 2 updates by [@&open-telemetry#8203;dependabot](https://github.com/dependabot) in [https://github.com/IBM/sarama/pull/2956](https://github.com/IBM/sarama/pull/2956) - chore(deps): bump github.com/eapache/go-resiliency from 1.6.0 to 1.7.0 by [@&open-telemetry#8203;dependabot](https://github.com/dependabot) in [https://github.com/IBM/sarama/pull/2944](https://github.com/IBM/sarama/pull/2944) - chore(deps): bump github.com/klauspost/compress from 1.17.8 to 1.17.9 by [@&open-telemetry#8203;dependabot](https://github.com/dependabot) in [https://github.com/IBM/sarama/pull/2926](https://github.com/IBM/sarama/pull/2926) ##### 🔧 Maintenance - fix(ci): correct docker-compose install by [@&open-telemetry#8203;dnwe](https://github.com/dnwe) in [https://github.com/IBM/sarama/pull/2954](https://github.com/IBM/sarama/pull/2954) ##### 📝 Documentation - fix(doc): correct JVM's config name corresponding to MaxWaitTime by [@&open-telemetry#8203;abhipranay](https://github.com/abhipranay) in [https://github.com/IBM/sarama/pull/2893](https://github.com/IBM/sarama/pull/2893) #### New Contributors - [@&open-telemetry#8203;abhipranay](https://github.com/abhipranay) made their first contribution in [https://github.com/IBM/sarama/pull/2893](https://github.com/IBM/sarama/pull/2893) - [@&open-telemetry#8203;kumakichi](https://github.com/kumakichi) made their first contribution in [https://github.com/IBM/sarama/pull/2909](https://github.com/IBM/sarama/pull/2909) - [@&open-telemetry#8203;richardartoul](https://github.com/richardartoul) made their first contribution in [https://github.com/IBM/sarama/pull/2939](https://github.com/IBM/sarama/pull/2939) **Full Changelog**: IBM/sarama@v1.43.2...v1.43.3 </details> --- ### Configuration 📅 **Schedule**: Branch creation - "on tuesday" (UTC), Automerge - At any time (no schedule defined). 🚦 **Automerge**: Disabled by config. Please merge this manually once you are satisfied. ♻ **Rebasing**: Whenever PR becomes conflicted, or you tick the rebase/retry checkbox. 🔕 **Ignore**: Close this PR and you won't be reminded about this update again. --- - [ ] If you want to rebase/retry this PR, check this box --- This PR was generated by [Mend Renovate](https://www.mend.io/free-developer-tools/renovate/). View the [repository job log](https://developer.mend.io/github/open-telemetry/opentelemetry-collector-contrib).  --------- Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com> Co-authored-by: opentelemetrybot <107717825+opentelemetrybot@users.noreply.github.com> Co-authored-by: Yang Song <songy23@users.noreply.github.com>

prestona and others added 2 commits July 23, 2024 18:30

prestona mentioned this pull request Jul 23, 2024

Re-ordering of offset commit requests can cause committed offset to move "backwards". #2940

Closed

puellanivis reviewed Jul 24, 2024

View reviewed changes

offset_manager.go Outdated Show resolved Hide resolved

prestona added 2 commits July 24, 2024 17:46

Improve clarity when returning a nil error value

08e09ad

Signed-off-by: Adrian Preston <PRESTONA@uk.ibm.com>

prestona mentioned this pull request Jul 25, 2024

Ensure ordering of offset commit requests #2941

Closed

puellanivis reviewed Jul 25, 2024

View reviewed changes

offset_manager_test.go Show resolved Hide resolved

dnwe added the fix label Aug 7, 2024

dnwe changed the title ~~Ensure ordering of offset commit requests (alternative version)~~ fix(consumer): maintain ordering of offset commit requests Aug 7, 2024

dnwe approved these changes Aug 7, 2024

View reviewed changes

dnwe merged commit 6b4f9be into IBM:main Aug 7, 2024
14 checks passed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix(consumer): maintain ordering of offset commit requests #2947

fix(consumer): maintain ordering of offset commit requests #2947

prestona commented Jul 23, 2024

dnwe left a comment

fix(consumer): maintain ordering of offset commit requests #2947

fix(consumer): maintain ordering of offset commit requests #2947

Conversation

prestona commented Jul 23, 2024

dnwe left a comment

Choose a reason for hiding this comment