Consumer group (Again)

README.md

@@ -24,24 +24,12 @@ KakfaEx supports the following Kafka features:
 * Fetch Messages
 * Message Compression with Snappy and gzip
 * Offset Management (fetch / commit / autocommit)
+* Consumer Groups
 
 See [Kafka Protocol Documentation](http://kafka.apache.org/protocol.html) and
  [A Guide to the Kafka Protocol](https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol)
 for details of these features.
 
-KafkaEx **does support** consumer groups for message consumption.  This feature
-was added in Kafka 0.8.2.  This translates to providing a consumer group
-name when committing offsets.  It is up to the client to assign partitions to
-workers in this mode of operation.
-
-KafkaEx currently provides **limited support** for the [Kafka ConsumerGroup
-API](https://cwiki.apache.org/confluence/display/KAFKA/A+Guide+To+The+Kafka+Protocol#AGuideToTheKafkaProtocol-GroupMembershipAPI)
-that was added in Kafka 0.9.0.  Most of the protocol requests are implemented
-in KafkaEx, but we do not yet support automatic joining and management of
-consumer group memebership (e.g., automatically assigning partitions to
-clients).  We are actively working on an implementation for automatic consumer
-group management.
-
 ## Using KafkaEx in an Elixir project
 
 The standard approach for adding dependencies to an Elixir application applies:
@@ -90,6 +78,35 @@ You can also override options when creating a worker, see below.
 
 ## Usage Examples
 
+### Consumer Groups
+
+To use a consumer group, first implement a handler module using
+`KafkaEx.GenConsumer`.
+
+```
+defmodule ExampleGenConsumer do
+  use KafkaEx.GenConsumer
+
+  alias KafkaEx.Protocol.Fetch.Message
+
+  require Logger
+
+  # note - messages are delivered in batches
+  def handle_message_set(message_set, state) do
+    for %Message{value: message} <- message_set do
+      Logger.debug(fn -> "message: " <> inspect(message) end)
+    end
+    {:async_commit, state}
+  end
+end
+```
+
+Then add a `KafkaEx.ConsumerGroup` to your application's supervision
+tree and configure it to use the implementation module.
+
+See the `KafkaEx.GenConsumer` and `KafkaEx.ConsumerGroup` documentation for
+details.
+
 ### Create a KafkaEx Worker
 
 KafkaEx worker processes manage the state of the connection to the Kafka broker.

config/config.exs

@@ -26,6 +26,11 @@ config :kafka_ex,
   max_restarts: 10,
   # Supervision max_seconds -  the time frame in which :max_restarts applies
   max_seconds: 60,
+  # Interval in milliseconds that GenConsumer waits to commit offsets.
+  commit_interval: 5_000,
+  # Threshold number of messages consumed for GenConsumer to commit offsets
+  # to the broker.
+  commit_threshold: 100,
   # This is the flag that enables use of ssl
   use_ssl: true,
   # see SSL OPTION DESCRIPTIONS - CLIENT SIDE at http://erlang.org/doc/man/ssl.html

lib/kafka_ex.ex

@@ -6,6 +6,12 @@ defmodule KafkaEx do
   alias KafkaEx.Protocol.ConsumerMetadata.Response, as: ConsumerMetadataResponse
   alias KafkaEx.Protocol.Fetch.Response, as: FetchResponse
   alias KafkaEx.Protocol.Fetch.Request, as: FetchRequest
+  alias KafkaEx.Protocol.Heartbeat.Request, as: HeartbeatRequest
+  alias KafkaEx.Protocol.Heartbeat.Response, as: HeartbeatResponse
+  alias KafkaEx.Protocol.JoinGroup.Request, as: JoinGroupRequest
+  alias KafkaEx.Protocol.JoinGroup.Response, as: JoinGroupResponse
+  alias KafkaEx.Protocol.LeaveGroup.Request, as: LeaveGroupRequest
+  alias KafkaEx.Protocol.LeaveGroup.Response, as: LeaveGroupResponse
   alias KafkaEx.Protocol.Metadata.Response, as: MetadataResponse
   alias KafkaEx.Protocol.Offset.Response, as: OffsetResponse
   alias KafkaEx.Protocol.OffsetCommit.Request, as: OffsetCommitRequest
@@ -14,6 +20,8 @@ defmodule KafkaEx do
   alias KafkaEx.Protocol.OffsetFetch.Request, as: OffsetFetchRequest
   alias KafkaEx.Protocol.Produce.Request, as: ProduceRequest
   alias KafkaEx.Protocol.Produce.Message
+  alias KafkaEx.Protocol.SyncGroup.Request, as: SyncGroupRequest
+  alias KafkaEx.Protocol.SyncGroup.Response, as: SyncGroupResponse
   alias KafkaEx.Server
   alias KafkaEx.Stream
 
@@ -75,6 +83,46 @@ defmodule KafkaEx do
     Server.call(worker, :consumer_group)
   end
 
+  @doc """
+  Sends a request to join a consumer group.
+  """
+  @spec join_group(JoinGroupRequest.t, Keyword.t) :: JoinGroupResponse.t
+  def join_group(request, opts \\ []) do
+    worker_name = Keyword.get(opts, :worker_name, Config.default_worker)
+    timeout = Keyword.get(opts, :timeout)
+    Server.call(worker_name, {:join_group, request, timeout}, opts)
+  end
+
+  @doc """
+  Sends a request to synchronize with a consumer group.
+  """
+  @spec sync_group(SyncGroupRequest.t, Keyword.t) :: SyncGroupResponse.t
+  def sync_group(request, opts \\ []) do
+    worker_name = Keyword.get(opts, :worker_name, Config.default_worker)
+    timeout = Keyword.get(opts, :timeout)
+    Server.call(worker_name, {:sync_group, request, timeout}, opts)
+  end
+
+  @doc """
+  Sends a request to leave a consumer group.
+  """
+  @spec leave_group(LeaveGroupRequest.t, Keyword.t) :: LeaveGroupResponse.t
+  def leave_group(request, opts \\ []) do
+    worker_name = Keyword.get(opts, :worker_name, Config.default_worker)
+    timeout = Keyword.get(opts, :timeout)
+    Server.call(worker_name, {:leave_group, request, timeout}, opts)
+  end
+
+  @doc """
+  Sends a heartbeat to maintain membership in a consumer group.
+  """
+  @spec heartbeat(HeartbeatRequest.t, Keyword.t) :: HeartbeatResponse.t
+  def heartbeat(request, opts \\ []) do
+    worker_name = Keyword.get(opts, :worker_name, Config.default_worker)
+    timeout = Keyword.get(opts, :timeout)
+    Server.call(worker_name, {:heartbeat, request, timeout}, opts)
+  end
+
   @doc """
   Return metadata for the given topic; returns for all topics if topic is empty string
 
@@ -197,7 +245,7 @@ defmodule KafkaEx do
     }, opts)
   end
 
-  @spec offset_commit(atom, OffsetCommitRequest.t) :: OffsetCommitResponse.t
+  @spec offset_commit(atom, OffsetCommitRequest.t) :: [OffsetCommitResponse.t]
   def offset_commit(worker_name, offset_commit_request) do
     Server.call(worker_name, {:offset_commit, offset_commit_request})
   end