feat: option to group metadata refreshes so only one is in-flight at a time by cupcicm · Pull Request #3225 · IBM/sarama

cupcicm · 2025-07-16T14:31:10Z

If other goroutines ask for another metadata refresh, we are going to accumulate all requested topics into a slice and we will queue another metadata request immediately after the current one is done.
It is possible to enable this behavior by setting config.Metadata.SingleFlight = true in your configuration.

I confirmed that this fixes #3224 in our setup.

metadata.go

client_test.go

puellanivis · 2025-07-17T15:25:53Z

client_test.go

+	var called atomic.Int32
+
+	seedBroker.setHandler(func(req *request) (res encoderWithHeader) {
+		require.EqualValues(t, 3, req.body.key(), "this test sends only Metadata requests")


Do not sure require here. The danger here is that we could be running this on a separate goroutine from the one running the test.

Note the condition on https://pkg.go.dev/[email protected]#T.Fatal

Also consider that using a regular if block is more than adequate:

if req.body.key() != 3 /* can we get a name in here instead? */ { t.Error("this test should only send Metadata requests") return }

Would you prefer that I used

if xxx { t.Error() }

or assert.EqualValues ? (rather than require. )

Usually the former is more direct and clear. In this case, the if cond { t.Error() } pattern can provide a more useful test failure message than just “3 != 3”.

client_test.go

puellanivis · 2025-07-17T16:16:22Z

metadata.go

+
+// Ongoing returns true if the refresh is ongoing.
+// You need to hold the lock to call this method.
+func (r *currentRefresh) Ongoing() bool {


This convenience function provides no convenience.

The currentRefresh lock must already be held, so we need to restrict this function from being called outside of this scope, so exporting it is not useful.

And since there is otherwise no locking, we could just export r.ongoing directly instead. Otherwise, we already can access r.ongoing ourselves.

You're right, sorry about this.

metadata.go

puellanivis · 2025-07-17T16:21:47Z

metadata.go

+			m.next.Unlock()
+			current.AddTopics(topics)
+			var ch = current.Start()
+			current.Unlock()


Requiring each code path to properly unlock this mutex explicitly is error prone.

Perhaps, we should compose all these functions into a ch, needToRetry := m.current.refreshTopics(topics)

❗ very nice !
I hadn't thought about this, let me do this it's going to be clearer 👍

metadata_test.go

puellanivis · 2025-07-17T16:34:28Z

metadata_test.go

+		// This one and the next one will be batched together because the first
+		// call is still ongoing when they run.
+		// So we will issue a single refresh call, for topic2 and topic3.
+		time.Sleep(10 * time.Millisecond)


I feel like we could use channels to gate execution guaranteeing that the sequence of steps taken are those that are expected, without needing to use an arbitrary wait time “just to be sure”.

first := make(chan struct{}) refresh := newSingleFlightRefresher(func(topics []string) error { select { case <-first: // already closed default: close(first) } // There should never be two refreshes that run concurrently, // so it's safe to change the map with no lock. for _, topic := range topics { called[topic]++ } time.Sleep(10 * time.Millisecond) // unsure how to guarantee the second and third refresh have looped. return nil }) var wg sync.WaitGroup wg.Add(1) go func() { defer wg.Done() assert.NoError(t, refresh([]string{"topic1"})) }() wg.Add(1) go func() { defer wg.Done() <-first assert.NoError(t, refresh([]string{"topic3", "topic4"})) }() wg.Add(1) go func() { defer wg.Done() <-first assert.NoError(t, refresh([]string{"topic3", "topic4"})) }() wg.Wait()

cupcicm · 2025-07-18T13:31:09Z

Thank you for your very thoughtful and very precise review @puellanivis
I have (I think) addressed all your comments except the naming one.

One possibly controversial change I made is the last commit, where I used the lower-level function you made me write (refreshOrQueue) in the test, which allows to resolve the race condition of "this request has been successfully enqueued but I haven't waited for the result to become available yet. If we allow ourselves to test this primitive (rather than the full Refresh()) we can test with no time.Sleep. If we don't, ... it's complicated (I think it's impossible because there is no point where we can assert we have enqueued the request without having to wait for the result).

About the naming of the configuration parameter, I wanted to discuss it with you. WDYT about

config.Metadata.BatchRequests
config.Metadata.DeduplicateRequests

? I personally liked "single flight" since I think it contains my idea of piggy backing on the previous refresh although it's a bit more complicated than usual singleflighting since we need to remember the topics.

puellanivis · 2025-07-21T08:31:43Z

I think it’s ok to use the more internal refreshOrQueue to remove race conditions, as long as the exported behavior relying upon that internal function is reasonably transparent. (Simple enough that we can just look at the code and see if it is valid or not.)

Unsure which new configuration parameter would be appropriate. config.Metadata.SingleFlight might not even be so bad, since my concern was mostly on just knowing that it applied to metadata requests only. 🤔 I probably would prefer less “deduplicate” as we are avoiding duplicate metadata refreshes, but we’re also merging in all outstanding topics together, not just removing duplicate ones. So maybe something to mean “batching” is not such a bad idea.

client_test.go

metadata.go

metadata_test.go

cupcicm

@puellanivis Would you prefer if the feature is called "config.Metadata.DeduplicateRequests" ?

client_test.go

cupcicm · 2025-07-21T09:40:39Z

metadata.go

+	}
+	if len(next.topics) > 0 {
+		r.AddTopics(next.topics)
+		next.topics = next.topics[:0]


OK, sorry. I am adding a clear() function to make it clearer what's happening.

metadata.go

metadata_test.go

dnwe · 2025-07-21T11:15:00Z

@puellanivis Would you prefer if the feature is called "config.Metadata.DeduplicateRequests" ?

@cupcicm I think SingleFlight is fine, as mentioned it is already grouped under Metadata and conveys “Use a single merged metadata request per flight (true) or parallel requests (false)” and is understood as evocative of https://pkg.go.dev/golang.org/x/sync/singleflight

After some testing I will probably flip the default to true before cutting the next release so the config would just remain as an opt-out anyway – if we even choose to offer it at all

cupcicm · 2025-07-22T09:49:47Z

OK. @puellanivis / @dnwe I have pushed an update that's exactly the same but with the signed commit, sorry.

I think we are good then? I believe I have addressed all your comments? Please tell me if that's not the case.

epot · 2025-07-28T07:03:43Z

@puellanivis / @dnwe what are we missing to be able to merge this? Thanks!

dnwe · 2025-07-28T08:40:20Z

@epot nothing blocking, just been on holiday for most of July. I intend to merge this week and then do some post-merge testing with the main branch at scale

epot · 2025-07-28T08:47:58Z

Ok great, thanks for the update!

puellanivis

Except for two minor concerns in the testing, everything else is just polishing.

PS: This means, once the test concerns are addressed, there’s no need to wait on any rereview.

client_test.go

metadata.go

metadata_test.go

epot · 2025-07-28T14:35:48Z

@puellanivis, I tried to address all your comments. Manu asked me to take over his PR while he was on holidays, but I am not necessarily super familiar with everything he did so I hope I didn't mess anything up. Could you take another look please?

puellanivis

Just a two things.

client_test.go

metadata_test.go

epot · 2025-07-29T13:43:18Z

@puellanivis I pushed again, let me know!

puellanivis · 2025-07-29T14:24:39Z

OK, I cannot see anything to comment about. 👍

epot · 2025-07-29T14:28:22Z

great! so what happens now?

dnwe · 2025-07-29T21:41:26Z

Please can you fix the DCO signoff?

And a single flight refresher implementation that fixes this issue: IBM#3224 Signed-off-by: Manu C <[email protected]> Signed-off-by: epot <[email protected]>

It defaults to false, and when you set it to true, sarama will use a single flight refresher to avoid sending multiple metadata requests at the same time. This fixes this issue: IBM#3224 Signed-off-by: Manu C <[email protected]> Signed-off-by: epot <[email protected]>

1. remove var = 2. use assert in goroutines 3. unexport a bunch of fields 4. some other minor things. Signed-off-by: Manu C <[email protected]> Signed-off-by: epot <[email protected]>

Rather than the Refresh method. It's a lower level tests, but it allows us to write a precise test with no time.Sleep. Signed-off-by: Manu C <[email protected]> Signed-off-by: epot <[email protected]>

Signed-off-by: Manu C <[email protected]> Signed-off-by: epot <[email protected]>

Signed-off-by: epot <[email protected]>

epot · 2025-07-30T07:06:52Z

Just did it @dnwe !

epot · 2025-07-31T07:01:39Z

I noticed you triggered more tests and one is failing. Should I look into it or is it known as flaky?

dnwe · 2025-07-31T07:09:58Z

Triggered a re-run to see

epot · 2025-07-31T07:49:37Z

Looks good now :)

dnwe

Thanks to all involved in both the debug, fix and review of this feature ❤️

Changes LGTM

epot · 2025-07-31T12:46:03Z

Great! Thanks :) Do you know when you plan to release it? Is there some place I can subscribe to know when it happens?

Enable #3225 by default (retaining the config option to permit an opt-out) Signed-off-by: Dominic Evans <[email protected]>

epot · 2025-08-12T08:13:26Z

@dnwe sorry for the ping, did you see my message above? I am curious when this is going to be released. Thanks!

dnwe · 2025-08-12T08:18:03Z

@epot no worries, yes it will be released soon, should be this week – with this and the ApiVersions adherence, there's a reasonable amount of change included so I have a few different teams doing some pre-release testing before I cut the tag

dnwe · 2025-08-12T08:20:56Z

Note also that you can go get github.com/IBM/sarama@main at any time to use the latest main branch version in your app's go.mod/go.sum so you can start using this level of code right away too

epot · 2025-08-12T08:48:35Z

Yes we are already doing that, but a bunch of customers where asking us when it would be released. Thanks a lot for the update!

cupcicm mentioned this pull request Jul 16, 2025

Metadata request burst #3224

Closed

airlock-warpstreamlabs bot force-pushed the cupcicm/singleflight_metadata_requests branch 2 times, most recently from 3a42eda to 1ea2b72 Compare July 16, 2025 14:37

ryanworl reviewed Jul 16, 2025

View reviewed changes

metadata.go Outdated Show resolved Hide resolved

airlock-warpstreamlabs bot force-pushed the cupcicm/singleflight_metadata_requests branch from 1ea2b72 to 6b3ce90 Compare July 16, 2025 14:53

puellanivis reviewed Jul 17, 2025

View reviewed changes

airlock-warpstreamlabs bot force-pushed the cupcicm/singleflight_metadata_requests branch 2 times, most recently from 5f60a69 to e224461 Compare July 18, 2025 13:15

puellanivis reviewed Jul 21, 2025

View reviewed changes

client_test.go Outdated Show resolved Hide resolved

client_test.go Outdated Show resolved Hide resolved

metadata.go Show resolved Hide resolved

metadata.go Outdated Show resolved Hide resolved

metadata_test.go Outdated Show resolved Hide resolved

cupcicm commented Jul 21, 2025

View reviewed changes

airlock-warpstreamlabs bot force-pushed the cupcicm/singleflight_metadata_requests branch from edfcadb to 716d30d Compare July 22, 2025 09:49

puellanivis reviewed Jul 28, 2025

View reviewed changes

client_test.go Outdated Show resolved Hide resolved

client_test.go Outdated Show resolved Hide resolved

metadata.go Outdated Show resolved Hide resolved

metadata.go Outdated Show resolved Hide resolved

metadata_test.go Outdated Show resolved Hide resolved

puellanivis reviewed Jul 29, 2025

View reviewed changes

client_test.go Outdated Show resolved Hide resolved

client_test.go Outdated Show resolved Hide resolved

metadata_test.go Outdated Show resolved Hide resolved

cupcicm added 5 commits July 30, 2025 09:04

Creates a "MetadataRefresh" interface.

1c946dc

And a single flight refresher implementation that fixes this issue: IBM#3224 Signed-off-by: Manu C <[email protected]> Signed-off-by: epot <[email protected]>

Cosmetic improvements.

32af561

1. remove var = 2. use assert in goroutines 3. unexport a bunch of fields 4. some other minor things. Signed-off-by: Manu C <[email protected]> Signed-off-by: epot <[email protected]>

Make the tests test the refreshOrQueue method.

11e1cdd

Rather than the Refresh method. It's a lower level tests, but it allows us to write a precise test with no time.Sleep. Signed-off-by: Manu C <[email protected]> Signed-off-by: epot <[email protected]>

Address comments

6e55b7b

Signed-off-by: Manu C <[email protected]> Signed-off-by: epot <[email protected]>

epot added 2 commits July 30, 2025 09:04

Address comments

50b34b5

Signed-off-by: epot <[email protected]>

fixes after review

4b96b6c

Signed-off-by: epot <[email protected]>

airlock-warpstreamlabs bot force-pushed the cupcicm/singleflight_metadata_requests branch from a16a5d6 to 4b96b6c Compare July 30, 2025 07:04

dnwe added the feat label Jul 30, 2025

dnwe changed the title ~~Adds a component that knows how to run a single concurrent metadata refresh~~ feat: option to group metadata refreshes so only one is in-flight at a time Jul 31, 2025

dnwe approved these changes Jul 31, 2025

View reviewed changes

dnwe merged commit e566ac6 into IBM:main Jul 31, 2025
21 of 22 checks passed

dnwe mentioned this pull request Aug 1, 2025

feat: use singleflight metadata by default #3231

Merged

dnwe added a commit that referenced this pull request Aug 6, 2025

feat: use singleflight metadata by default (#3231)

79f54aa

Enable #3225 by default (retaining the config option to permit an opt-out) Signed-off-by: Dominic Evans <[email protected]>

puellanivis mentioned this pull request Aug 14, 2025

All partitions of a topic execute "retring message" and "refreshmatedata" when their broker is closed #3261

Closed

This was referenced Mar 3, 2026

Upgrade elastic/sarama fork to v1.46.x to include Metadata.SingleFlight elastic/beats#49210

Closed

Update fork baseline from to v1.46.x elastic/sarama#30

Closed

Conversation

cupcicm commented Jul 16, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

cupcicm commented Jul 18, 2025

Uh oh!

puellanivis commented Jul 21, 2025

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

cupcicm left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

dnwe commented Jul 21, 2025

Uh oh!

cupcicm commented Jul 22, 2025

Uh oh!

epot commented Jul 28, 2025

Uh oh!

dnwe commented Jul 28, 2025

Uh oh!

epot commented Jul 28, 2025

Uh oh!

puellanivis left a comment • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

epot commented Jul 28, 2025

Uh oh!

puellanivis left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

epot commented Jul 29, 2025

Uh oh!

puellanivis commented Jul 29, 2025

Uh oh!

epot commented Jul 29, 2025

Uh oh!

dnwe commented Jul 29, 2025

Uh oh!

puellanivis left a comment •

edited

Loading