Add Client.Coordinator() to retrieve the coordinating broker for a consumer group #411

wvanbergen · 2015-04-09T09:30:56Z

This is the first step towards consumer group offset management, and later full consumer group management in Kafka 0.8.3.

Careful review of the 2 locks would be appreciated. Do we need to locks or can be just use the single lock for everything?

Extracted from #379.

@Shopify/kafka

wvanbergen · 2015-04-09T09:31:24Z

client.go

+		return coordinator, nil
+	}
+
+	coordinator, err := client.refreshCoordinator(consumerGroup, 10)


Number of retries is hardcoded for now.

wvanbergen · 2015-04-09T09:35:09Z

Outstanding question: in the functional test, the retry attempts and backoff are set to very high values right now. Normally the default values are more than enough, but during the first call to Coordinator, Kafka actually creates the __consumer_offsets topic, which can take a long time. Any subsequent call to the function will be a lot faster.

I am not sure if this warrants using different configuration values. On Travis CI, this initial call takes about 4 seconds.

wvanbergen · 2015-04-09T10:07:15Z

Note: this only works against Kafka 0.8.2 and higher.

eapache · 2015-04-09T14:47:43Z

For simplicity, let's stick to just one lock for now. We can add fine-grained locking later if it turns out to be a performance bottleneck, but I doubt it.
Is there a reason not to follow the same pattern as Leader here, with a cachedCoordinator function that returns the (opened) broker and a refreshCoordinator function that just updates the maps and doesn't return the broker directly?
For retries, perhaps it is worth detecting the "need to create the topic" case (what error is that?) and adding a one-time separately configurable back-off of 5 seconds for that case, and otherwise using the metadata retries/backoffs. I am alternatively OK with just creating entirely separate retry/backoff configs for this path.

wvanbergen · 2015-04-09T14:59:09Z

will do.
I think this is simpler and separates concerns better. refreshCoordinator just cares about executing the request and has no side effects, and doesn't have to care about cache. Coordinator deals with the cache.
Let me investigate what happens when you create a second consumergroup. If my understanding is correct, that will return the same error (ErrConsumerCoordinatorNotAvailable), but it will be functional much faster because it doesn't need to create the topic. I will report back later.

eapache · 2015-04-09T15:04:34Z

It just seems weird to have the broker cache updated in the refresh method but the coordinator cache updated in the Coordinator method - I feel like those updates should live together.

The concern I separated with the Leader design is lazy open - only cachedLeader has to worry about opening the broker, since it is the only thing that returns a broker.

Also, nothing uses the invalidateCoordinator method?

wvanbergen · 2015-04-09T15:06:26Z

The invalidateCoordinator will be used by methods that use the coordinator and figure out that it actually is not the correct one.

eapache · 2015-04-09T15:07:34Z

The invalidateCoordinator will be used by methods that use the coordinator and figure out that it actually is not the correct one.

No no no, this is the same mistake I made with disconnectBroker :(

eapache · 2015-04-09T15:08:24Z

client.go


 	// If the number of partitions is large, we can get some churn calling cachedPartitions,
 	// so the result is cached.  It is important to update this value whenever metadata is changed
 	cachedPartitionsResults map[string][maxPartitionIndex][]int32
-	lock                    sync.RWMutex // protects access to the maps, only one since they're always written together
+
+	lock sync.RWMutex // protects access to the metadata and coordinator maps


and broker map

and cachedPartitionsResults map

wvanbergen · 2015-04-09T15:19:43Z

Updated refreshCoordinator to never register or open the broker, Coordinator now always does these tasks.

wvanbergen · 2015-04-09T15:21:30Z

Remind me again why that caused issues for disconnectBroker?

wvanbergen · 2015-04-09T15:56:41Z

[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_0 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_0 is #9094 (localhost:9094).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9094
[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_1 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_1 is #9095 (localhost:9095).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9095
[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_2 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_2 is #9091 (localhost:9091).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9091
[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_3 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_3 is #9092 (localhost:9092).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9092
[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_4 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_4 is #9093 (localhost:9093).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9093
[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_5 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_5 is #9094 (localhost:9094).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9094
[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_6 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_6 is #9095 (localhost:9095).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9095
[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_7 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_7 is #9091 (localhost:9091).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9091
[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_8 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_8 is #9092 (localhost:9092).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9092
[sarama] 2015/04/09 15:53:45 client/coordinator Requesting coordinator for consumergoup another_new_consumer_group_9 from localhost:9091.
[sarama] 2015/04/09 15:53:45 client/coordinator Coordinator for consumergoup another_new_consumer_group_9 is #9093 (localhost:9093).
[sarama] 2015/04/09 15:53:45 Connected to broker localhost:9093

Looks like for all subsequent attempts for new consumergroups, the call succeeds immediately. So ErrConsumerCoordinatorNotAvailable is only returned when the topic is being created. A separate longer sleep call when this happens sounds like a good idea. 👍

eapache · 2015-04-09T16:10:03Z

Remind me again why that caused issues for disconnectBroker?

#15
#23 (comment)

Basically I made it private but used it from outside the client. The semantics were confused. Lots of things.

The solution finally boiled down to two different tasks the user might want to accomplish:

bounce a broker connection if e.g. it's timed out
refresh metadata if the leadership has moved

The first can already be accomplished (now that we have lazy connections) by just calling Close() on the broker and asking for it again. The second can be accomplished (in the leader case) by calling RefreshMetadata().

So based on all that, I think we should add RefreshCoordinator(topic) error as a public client method (and get rid of invalidateCoordinator).

edit, summary: invalidateCoordinator, like disconnectBroker is just a way to flag "please do a refresh the next time I ask for this broker". It is simpler and more powerful and easier to grok if we just expose a method that is "do the refresh directly"

wvanbergen · 2015-04-09T17:14:14Z

Removed invalidateCoordinator, added RefreshCoordinator.

wvanbergen · 2015-04-10T11:26:27Z

Some final 👀 ?

eapache · 2015-04-10T12:50:42Z

client.go

@@ -42,6 +43,14 @@ type Client interface {
 	// offset, OffsetNewest for the offset of the message that will be produced next, or a time.
 	GetOffset(topic string, partitionID int32, time int64) (int64, error)

+	// Coordinator returns the coordinating broker for a consumer group. It will return a locally cached
+	// value if it's available. This may be stale, in which case you should call RefreshCoordinator


I don't like the UX of saying "this may be stale", I think the necessary information is conveyed with just "it's locally cached".

eapache · 2015-04-10T16:46:44Z

client.go

+		return err
+	}
+
+	coordinator := &Broker{


once #413 lands this can go away

eapache · 2015-04-10T17:00:22Z

client.go

@@ -42,6 +42,14 @@ type Client interface {
 	// offset, OffsetNewest for the offset of the message that will be produced next, or a time.
 	GetOffset(topic string, partitionID int32, time int64) (int64, error)

+	// Coordinator returns the coordinating broker for a consumer group. It will return a locally cached
+	// value if it's available. This value is cached locally; you can call RefreshCoordinator to update


you don't need to says it's cached locally twice :)

wvanbergen · 2015-04-10T20:42:26Z

@eapache final 👀

eapache · 2015-04-10T21:02:40Z

client.go

+// registerBroker makes sure a broker received by a Metadata or Coordinator request is registered
+// in the brokers map. It returns the broker that is registered, which may be the provided broker,
+// or a previously registered Broker instance.
+func (client *client) registerBroker(broker *Broker) {


maybe worth mentioning this must be called with the lock held in the comment?

eapache · 2015-04-10T21:13:02Z

Very minor nits, but LGTM. Squash it and

… retrieve the coordinating broker for a consumer group.

Add Client.Coordinator() to retrieve the coordinating broker for a consumer group

zaynetro · 2015-05-22T06:44:04Z

Note: this only works against Kafka 0.8.2 and higher.

Damn, I spent hours trying to get Coordinator to work on v0.8.1.1. Maybe, it should be added to the docs?

wvanbergen · 2015-05-22T12:25:39Z

@zaynetro #453

zaynetro · 2015-05-22T12:32:17Z

Great!

wvanbergen reviewed Apr 9, 2015
View reviewed changes

wvanbergen force-pushed the coordinator branch 3 times, most recently from 300b948 to 9596184 Compare April 9, 2015 10:03

eapache reviewed Apr 9, 2015
View reviewed changes

wvanbergen force-pushed the coordinator branch from 0a796ab to 5f3ea66 Compare April 9, 2015 15:39

wvanbergen force-pushed the coordinator branch from e1d6bcb to 5720591 Compare April 9, 2015 16:00

wvanbergen force-pushed the coordinator branch from 79953ea to e691ae5 Compare April 9, 2015 17:19

eapache reviewed Apr 10, 2015
View reviewed changes

wvanbergen force-pushed the coordinator branch from 971d2cd to 86f217a Compare April 10, 2015 16:31

eapache reviewed Apr 10, 2015
View reviewed changes

client.go

return err

}

coordinator := &Broker{

Copy link

Contributor

eapache Apr 10, 2015

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

once #413 lands this can go away

wvanbergen force-pushed the coordinator branch from a6cf6c0 to 2e0eaca Compare April 10, 2015 16:59

eapache reviewed Apr 10, 2015
View reviewed changes

wvanbergen force-pushed the coordinator branch from 6857713 to 7ce3dcb Compare April 10, 2015 19:22

eapache reviewed Apr 10, 2015
View reviewed changes

Add Client.Coordinator(topic) and Client.RefreshCoordinator(topic) to…

95649f5

… retrieve the coordinating broker for a consumer group.

wvanbergen force-pushed the coordinator branch from 87ab777 to 95649f5 Compare April 10, 2015 21:18

wvanbergen added a commit that referenced this pull request Apr 10, 2015

Merge pull request #411 from Shopify/coordinator

34f21c8

Add Client.Coordinator() to retrieve the coordinating broker for a consumer group

wvanbergen merged commit 34f21c8 into master Apr 10, 2015

wvanbergen deleted the coordinator branch April 10, 2015 21:26

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add Client.Coordinator() to retrieve the coordinating broker for a consumer group #411

Add Client.Coordinator() to retrieve the coordinating broker for a consumer group #411

wvanbergen commented Apr 9, 2015

wvanbergen Apr 9, 2015

wvanbergen commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

eapache commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

eapache commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

eapache commented Apr 9, 2015

eapache Apr 9, 2015

eapache Apr 9, 2015

wvanbergen commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

eapache commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

wvanbergen commented Apr 10, 2015

eapache Apr 10, 2015

eapache Apr 10, 2015

eapache Apr 10, 2015

wvanbergen commented Apr 10, 2015

eapache Apr 10, 2015

eapache commented Apr 10, 2015

zaynetro commented May 22, 2015

wvanbergen commented May 22, 2015

zaynetro commented May 22, 2015

Add Client.Coordinator() to retrieve the coordinating broker for a consumer group #411

Add Client.Coordinator() to retrieve the coordinating broker for a consumer group #411

Conversation

wvanbergen commented Apr 9, 2015

wvanbergen Apr 9, 2015

Choose a reason for hiding this comment

wvanbergen commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

eapache commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

eapache commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

eapache commented Apr 9, 2015

eapache Apr 9, 2015

Choose a reason for hiding this comment

eapache Apr 9, 2015

Choose a reason for hiding this comment

wvanbergen commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

eapache commented Apr 9, 2015

wvanbergen commented Apr 9, 2015

wvanbergen commented Apr 10, 2015

eapache Apr 10, 2015

Choose a reason for hiding this comment

eapache Apr 10, 2015

Choose a reason for hiding this comment

eapache Apr 10, 2015

Choose a reason for hiding this comment

wvanbergen commented Apr 10, 2015

eapache Apr 10, 2015

Choose a reason for hiding this comment

eapache commented Apr 10, 2015

zaynetro commented May 22, 2015

wvanbergen commented May 22, 2015

zaynetro commented May 22, 2015