Implement Kafka 0.9 protocol additions #577

wvanbergen · 2015-12-07T14:33:06Z

@eapache Am I on the right track here?

wvanbergen · 2015-12-07T14:37:00Z

join_group_request.go

+type GroupProtocol interface {
+	encodeGroupProtocol(packetEncoder) error
+	decodeGroupProtocol(packetDecoder) error
+}


This part is a bit tricky; the protocol defines this to be extensible based on the ProtocolType. It continues to only define the specifics for the "consumer" protocol.

I am tempted to just make this []byte and handle the consumer-specific implementation via Encoder interface? Not sure, but I dislike a public interface with no public methods.

edenhill · 2015-12-07T15:06:09Z

errors.go

@@ -37,6 +37,9 @@ var ErrShuttingDown = errors.New("kafka: message received by producer in process
 // ErrMessageTooLarge is returned when the next message to consume is larger than the configured Consumer.Fetch.Max
 var ErrMessageTooLarge = errors.New("kafka: message is larger than Consumer.Fetch.Max")

+// ErrMessageTooLarge is returned when a JoinGroup request returns a protocol type that is not supported by sarama.
+var ErrUnknownGroupProtocol = errors.New("kafka: encountered an unknown group protocol")


This can't happen, the broker selects the protocols supported by all members, and if no such set exists throws a InconsistentGroupProtocol

Right now this error is only used when the ProtocolType on the request is set (by the user) to an unknown value (i.e. anything but consumer).

eapache · 2015-12-07T15:07:04Z

You're going to need to add all the new KError values to this as well.

eapache · 2015-12-07T15:08:45Z

join_group_request.go

+}
+
+func (r *JoinGroupRequest) decode(pd packetDecoder) (err error) {
+	r.GroupId, err = pd.getString()


these can all be collapsed into their if lines because err is predefined

wvanbergen · 2015-12-07T16:24:11Z

Added new KErrors, and Heartbeat request and response pair.

eapache · 2015-12-07T16:43:08Z

errors.go

+	case ErrGroupAuthorizationFailed:
+		return "kafka server: The client is not authorized to access this group"
+	case ErrClusterAuthorizationFailed:
+		return "kafka server: The client is not authorized to send this request type"


for consistency, the existing errors end in punctuation

wvanbergen · 2015-12-07T18:50:11Z

I've implemented all the new response/request pairs. I have also removed the ConsumerGroup specialization stuff and left it at just byte-arrays. We can later decide how to handle that best.

eapache · 2015-12-07T18:56:38Z

LGTM. Collapse all the decode if statements and add tests, then

wvanbergen · 2015-12-08T16:53:37Z

join_group_request_test.go

+		0, 3, 't', 'w', 'o', // Protocol name
+		0, 0, 0, 3, 0x04, 0x05, 0x06, // protocol metadata
+		0, 3, 'o', 'n', 'e', // Protocol name
+		0, 0, 0, 3, 0x01, 0x02, 0x03, // protocol metadata


This gets assembled from a map, so the order is not deterministic. @eapache how have you gotten around this elsewhere?

Only test the 0/1 cases :P

wvanbergen · 2015-12-08T22:47:59Z

I changed some slice types into maps to make the types easier to use, and added tests to all the things. This is ready for final review.

wvanbergen · 2015-12-09T13:04:41Z

Note: some messages, fields, and errors are also renamed as part of the 0.9 release. For backwards compatibility reasons, these are not included.

eapache · 2015-12-09T14:49:48Z

describe_groups_response.go

+package sarama
+
+type DescribeGroupsResponse struct {
+	Groups []*GroupDescription


kinda/almost/sorta makes sense for this to be a map keyed by group ID, but not sure if it's worth it at this point

it would make decode rather more complex

I also like that this matches the request: you ask for a slice of groups, you get a slice of group descriptions back.

eapache · 2015-12-09T14:54:33Z

Implement Kafka 0.9 protocol additions

wvanbergen reviewed Dec 7, 2015
View reviewed changes

wvanbergen force-pushed the kafka_09_protocol branch from c48c940 to 7a36173 Compare December 7, 2015 15:00

edenhill reviewed Dec 7, 2015
View reviewed changes

eapache reviewed Dec 7, 2015
View reviewed changes

wvanbergen force-pushed the kafka_09_protocol branch from dce94b6 to 821ca5b Compare December 7, 2015 16:02

wvanbergen changed the title ~~[WIP] Implement new Group protocol messages.~~ [WIP] Implement Kafka 0.9 protocol additions Dec 7, 2015

wvanbergen force-pushed the kafka_09_protocol branch from 07eec68 to 18a7ed2 Compare December 7, 2015 16:24

eapache reviewed Dec 7, 2015
View reviewed changes

Add new protocol errors with their respective messages.

2bac783

wvanbergen force-pushed the kafka_09_protocol branch from 3bb8696 to 21f7755 Compare December 7, 2015 18:49

wvanbergen force-pushed the kafka_09_protocol branch 2 times, most recently from 39e4207 to 611688b Compare December 8, 2015 14:15

wvanbergen reviewed Dec 8, 2015
View reviewed changes

wvanbergen force-pushed the kafka_09_protocol branch from 611688b to 9a8900b Compare December 8, 2015 21:13

wvanbergen added 9 commits December 8, 2015 17:06

Add JoinGroup request and response pair.

1badb8e

Add Heartbeat request and response pair.

b1d40bd

Add LeaveGroup request and response pair.

ccec1aa

Add SyncGroup request and response pair.

0d36de6

Add ListGroups request and response pair.

037310a

Add DescribeGroup request and response pair.

7611ea6

Register new request type keys in request.go

a050648

Better assertion failure messages from request_test.

9dbdf89

Add tests for all new request types.

1621c49

wvanbergen force-pushed the kafka_09_protocol branch from 9a8900b to cacc655 Compare December 8, 2015 22:45

wvanbergen changed the title ~~[WIP] Implement Kafka 0.9 protocol additions~~ Implement Kafka 0.9 protocol additions Dec 8, 2015

Add tests for all new response types

756801d

wvanbergen force-pushed the kafka_09_protocol branch from cacc655 to 756801d Compare December 8, 2015 22:49

eapache reviewed Dec 9, 2015
View reviewed changes

wvanbergen added a commit that referenced this pull request Dec 9, 2015

Merge pull request #577 from Shopify/kafka_09_protocol

159e999

Implement Kafka 0.9 protocol additions

wvanbergen merged commit 159e999 into master Dec 9, 2015

wvanbergen deleted the kafka_09_protocol branch December 9, 2015 15:03

eapache mentioned this pull request Dec 9, 2015

Implement new kafka 0.9 protocols #574

Closed

mattes mentioned this pull request Jan 20, 2016

Consumer Groups/ Group Membership API in 0.9.x #597

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Implement Kafka 0.9 protocol additions #577

Implement Kafka 0.9 protocol additions #577

wvanbergen commented Dec 7, 2015

wvanbergen Dec 7, 2015

eapache Dec 7, 2015

edenhill Dec 7, 2015

wvanbergen Dec 7, 2015

eapache commented Dec 7, 2015

eapache Dec 7, 2015

wvanbergen commented Dec 7, 2015

eapache Dec 7, 2015

wvanbergen commented Dec 7, 2015

eapache commented Dec 7, 2015

wvanbergen Dec 8, 2015

eapache Dec 8, 2015

wvanbergen commented Dec 8, 2015

wvanbergen commented Dec 9, 2015

eapache Dec 9, 2015

eapache Dec 9, 2015

wvanbergen Dec 9, 2015

eapache Dec 9, 2015

eapache commented Dec 9, 2015

Implement Kafka 0.9 protocol additions #577

Implement Kafka 0.9 protocol additions #577

Conversation

wvanbergen commented Dec 7, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eapache commented Dec 7, 2015

Choose a reason for hiding this comment

wvanbergen commented Dec 7, 2015

Choose a reason for hiding this comment

wvanbergen commented Dec 7, 2015

eapache commented Dec 7, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wvanbergen commented Dec 8, 2015

wvanbergen commented Dec 9, 2015

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

eapache commented Dec 9, 2015