Consumer group (Again) #219

dantswain · 2017-07-25T21:36:38Z

This is my continuation of #207 which is in turn a continuation of #195 .

It's still WIP - feel free to review as it goes, but I'll post when done.

sourcelevel-bot · 2017-07-25T21:37:17Z

lib/kafka_ex/consumer_group.ex

+    join_response = %JoinGroupResponse{error_code: :no_error} =
+      KafkaEx.join_group(join_request, worker_name: worker_name, timeout: session_timeout + @session_timeout_padding)
+
+    Logger.debug("Joined consumer group #{group_name}")


Prefer lazy Logger calls.

sourcelevel-bot · 2017-07-25T21:37:18Z

lib/kafka_ex/consumer_group.ex

+        {:ok, new_state}
+
+      %HeartbeatResponse{error_code: :rebalance_in_progress} ->
+        Logger.debug("Rebalancing consumer group #{group_name}")


Prefer lazy Logger calls.

sourcelevel-bot · 2017-07-25T21:37:18Z

lib/kafka_ex/consumer_group.ex

+
+    %LeaveGroupResponse{error_code: :no_error} = KafkaEx.leave_group(leave_request, worker_name: worker_name)
+
+    Logger.debug("Left consumer group #{group_name}")


Prefer lazy Logger calls.

sourcelevel-bot · 2017-07-25T21:37:18Z

lib/kafka_ex/gen_consumer.ex

+    [%OffsetCommitResponse{topic: ^topic, partitions: [^partition]}] =
+      KafkaEx.offset_commit(worker_name, request)
+
+    Logger.debug("Committed offset #{topic}/#{partition}@#{offset} for #{group}")


Prefer lazy Logger calls.

i.e., instead of `handle_message/2` we have `handle_message_set/2` It's very easy for the user to iterate over the batch, and there is at least one use case for passing batches to the consumer: parallelized processing of messages. At Simpli.fi we have some applications where each message we receive from Kafka some amount of work that can be done asynchronously, and we've found a big performance increase by processing those in parallel. There are some delivery guarantee implications to that, but it should be up to the user to decide how to handle that.

Mostly just line length. Added an intermediate variable in a couple places to handle long case statement input.

This makes more intuitive sense - one wouldn't really start the manager process without the supervisor.

GenConsumer doesn't necessarily need to know anything about the way partitions are assigned - it makes more sense for the ConsumerGroup code to deal with this. I made it an optional callback that defaults to the round robin algorithm. I tried to update the docs where applicable.

* Group the gen_server_opts * Specify heartbeat and session timeout opts

I tried to add a test for this but had significant trouble getting the test to work. I BELIEVE the issues with the test are because of the number of connects/disconnects happening in rapid succession, but I'm not sure and I'd like to revisit that later.

This should be more robust and exercises some of the ops functions

Cleaned up the tests a bit

dantswain · 2017-08-08T00:10:27Z

@bjhaid @joshuawscott I think this is finally ready for review. There are a couple things I might like to clean up later, but they can be in follow-up PRs. I mostly wanted to nail down the API first.

@dcuddeback If you have time, I would greatly appreciate any review you can give here. This is still mostly your vision, but I changed a handful of things. This may not be an exhaustive list:

Renamed :ack / :commit to :async_commit and :sync_commit, respectively.
GenConsumer handles message sets instead of individual messages now. @joshuawscott and I have found this to be useful in some applications (e.g., we can process messages in parallel), and it's trivial for a user to iterate over the message set on their own if they want to.
ConsumerGroup is now the supervisor and it starts a ConsumerGroup.Manager process. I thought this made more sense from an API standpoint - ConsumerGroup.start_link is now the fundamental entry point for a consumer group.
Remove partition assignment responsibility from GenConsumer - this isn't really part of consuming messages. Round-robin is probably fine for most users. Users still have the ability to specify a partition assignment callback in ConsumerGroup.start_link.
Gave the documentation a relatively thorough editing.
Added tests for the consumer group.
Added a bunch of ops-friendly functions.

bjhaid · 2017-08-08T00:52:06Z

ack, this is some massive diff, I'll find sometime to review it before the end of the week

joshuawscott · 2017-08-08T18:29:10Z

lib/kafka_ex/consumer_group/manager.ex

+      session_timeout: session_timeout,
+    }
+
+    join_response = %JoinGroupResponse{error_code: :no_error} =


This will crash if we get any error back from Kafka. I don't know if it's necessarily good to crash within a private function that's in an essentially private module. It will make it difficult to diagnose for a library user. Maybe we should crash with a helpful message here (like display a human-readable version of the error code that we got)

I think what I'd like to do here is follow up on some of these with separate PRs. I'll make issues for them.

This one is #223

joshuawscott · 2017-08-08T19:50:50Z

test/integration/consumer_group_implementation_test.exs

+    assignments2 = ConsumerGroup.assignments(context[:consumer_group_pid2])
+    assert 2 == length(assignments1)
+    assert 2 == length(assignments2)
+    refute assignments1 == assignments2


maybe assert that none of the assignments in assignments1 are in assignments2

joshuawscott

Looks good to me!

dantswain · 2017-08-11T00:33:27Z

@bjhaid @dcuddeback ping. Sorry for the size of the PR :(

bjhaid

LGTM

dcuddeback and others added 12 commits May 10, 2017 11:05

implements consumer groups

a88ea57

improves error handling

3cb335e

makes session timeout padding a module attribute

da7ce51

refactor to combine common code in sync group functions

6c279d2

attempt to make ConsumerGroup logic easier to follow

f77c679

replace GenServer timeout with timer

32fd4e5

adds Metadata.Response.partitions_for_topic/2

6f40c7d

manage dynamic assignments with ConsumerGroup.Supervisor

7609388

improve readability of deeply-nested GenConsumer.Supervisor.start_link/4

14793e3

starts KafkaEx.Server per consumer

9d0b4cc

Merge branch 'master' into consumer-group

799dcc3

Unbreak merge issues

46a8054

sourcelevel-bot bot reviewed Jul 25, 2017

View reviewed changes

dantswain mentioned this pull request Jul 25, 2017

Consumer Group #207

Closed

dantswain added 13 commits July 26, 2017 10:55

Merge branch 'master' into consumer-group

407dffb

Thunk logger calls

d16c15c

Move supervisors to subdirectories

eb3e1e3

Rename :ack / :commit to :async_commit / :sync_commit

bbf2695

Merge branch 'master' into consumer-group

013774f

Merge branch 'master' into consumer-group

e3926aa

Strict credo cleanup

69eba79

Mostly just line length. Added an intermediate variable in a couple places to handle long case statement input.

Edit documentation

3f97193

Refactor st ConsumerGroup is the supervisor

90d93b8

This makes more intuitive sense - one wouldn't really start the manager process without the supervisor.

Fix documentation oversight

eff912f

Fix default worker name for test helper

9c32696

Create topic on docker up

3a7719c

dantswain added 16 commits August 1, 2017 15:33

Test that exiting consumer also commits offsets

2dafc01

Test across all partitions on the topic

52dcba9

Add a second consumer to the test group

74036e2

Refactor TestObserver

15bf18f

Better assignment tests

8a0cd14

Fix test command for travis

8818e6a

Assert all 4 handlers

1f29f41

Test for rebalancing

b3e73c0

Factoring/documentation for manager opts

d15e82b

* Group the gen_server_opts * Specify heartbeat and session timeout opts

Clean up test, Supervisor usage

dde8f7d

Add note about multiple topics

48cf82f

I tried to add a test for this but had significant trouble getting the test to work. I BELIEVE the issues with the test are because of the number of connects/disconnects happening in rapid succession, but I'm not sure and I'd like to revisit that later.

Add several ops helper functions

e122f4c

Test assignments directly

ae4b964

This should be more robust and exercises some of the ops functions

Add ability for consumer to handle messages

d25562b

Cleaned up the tests a bit

More test cleanup

ede2b09

dantswain requested review from bjhaid and joshuawscott August 8, 2017 00:02

joshuawscott reviewed Aug 8, 2017

View reviewed changes

joshuawscott approved these changes Aug 8, 2017

View reviewed changes

dantswain mentioned this pull request Aug 9, 2017

Consumer Group - Meaningful error on JoinGroupResponse error #223

Closed

Review feedback

ca7ff34

bjhaid approved these changes Aug 19, 2017

View reviewed changes

dantswain merged commit b0fef32 into master Aug 29, 2017

dantswain deleted the consumer-group branch August 29, 2017 13:04

dantswain mentioned this pull request Aug 29, 2017

Update consumer group API to support kafka 0.9.0 #67

Closed


		%LeaveGroupResponse{error_code: :no_error} = KafkaEx.leave_group(leave_request, worker_name: worker_name)

		Logger.debug("Left consumer group #{group_name}")

Consumer group (Again) #219

Consumer group (Again) #219

Uh oh!

Conversation

dantswain commented Jul 25, 2017

Uh oh!

sourcelevel-bot bot Jul 25, 2017

Choose a reason for hiding this comment

Uh oh!

sourcelevel-bot bot Jul 25, 2017

Choose a reason for hiding this comment

Uh oh!

sourcelevel-bot bot Jul 25, 2017

Choose a reason for hiding this comment

Uh oh!

sourcelevel-bot bot Jul 25, 2017

Choose a reason for hiding this comment

Uh oh!

dantswain commented Aug 8, 2017

Uh oh!

bjhaid commented Aug 8, 2017

Uh oh!

joshuawscott Aug 8, 2017

Choose a reason for hiding this comment

Uh oh!

dantswain Aug 9, 2017

Choose a reason for hiding this comment

Uh oh!

joshuawscott Aug 8, 2017

Choose a reason for hiding this comment

Uh oh!

joshuawscott left a comment

Choose a reason for hiding this comment

Uh oh!

dantswain commented Aug 11, 2017

Uh oh!

bjhaid left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants