 defmodule KafkaExExamples do
   @moduledoc """
-  Documentation for KafkaExExamples.
+  Main supervisor module
+
+  This could be your app's main module that has its `Application.start/2`
+  callback, or it could be a supervisor implementation inside your project.
+
+  The only part of this that is specific to KafkaEx is the consumer group
+  setup within the `start/2` function and the `assign_partitions/2` function
+  which could be implemented anywhere.
   """
 
   defmodule KafkaExExamples.Consumer do
+    @moduledoc """
+    Main consumer implementation
+
+    This implementsa KafkaEx.GenConsumer.  The only function we need to
+    implement is `handle_message_set/2`.
+    """
+
+    # this is required
     use KafkaEx.GenConsumer
 
+    # this is not required - I have aliased it for convenience
     alias KafkaEx.Protocol.Fetch.Message
 
+    # this is not generally required, but is required here so that we can
+    # log messages
     require Logger
 
+    @doc """
+    Main message handling callback function
+
+    Note that we receive multiple messages each time.  Because of the way that
+    Kafka works, each request to the broker can provide us with multiple
+    messages.
+
+    We return `{:async_commit, consumer_state}` - we do not modify the state
+    and we want offsets to be committed asynchronously.  Async commits
+    balance safety and performance.
+    """
+    @spec handle_message_set([Message.t], term) :: {:async_commit, term}
     def handle_message_set(messages, consumer_state) do
+      # just loop through each message and print it out
       for  message = %Message{} <- messages do
         Logger.debug(fn -> "GOT: #{inspect message}" end)
       end

    end
  end

  use Application

  alias KafkaEx.ConsumerGroup.PartitionAssignment
  alias KafkaExExamples.Consumer

  require Logger

  def start(_type, _args) do
    import Supervisor.Spec

    consumer_group_opts = [
      commit_interval: 1_000,
      heartbeat_interval: 1_000,
      name: ExampleConsumerGroup,
      gen_server_opts: [name: ExampleConsumerGroup.Manager],
      partition_assignment_callback: &assign_partitions/2,
      # must be >= group.min.session.timeout.ms from broker config
      session_timeout: 6_000
    ]

    children = [
      supervisor(
        KafkaEx.ConsumerGroup,

    Supervisor.start_link(children, supervisor_opts)
  end

  @doc """
  Partition assignment callback

  This is optional.  We choose to override it here so that we can use it as a
  hook to log a message.  This could be used for, e.g., collecting metrics in
  a production system.
  """
  def assign_partitions(members, partitions) do
    # call through to the default KafkaEx partition assignment algorithm
    result = PartitionAssignment.round_robin(members, partitions)

    # log out the assignments for demonstration purposes
    Logger.debug(fn ->
      "ASSIGN: #{inspect members} | #{inspect partitions} | #{inspect result}"
    end)

    # return the assignment result
    result
  end
end
