Enable graceful shutdown of a GenConsumer

Add a user-configurable :shutdown to GenConsumer's child_spec.

Addresses kafkaex#434.

lib/kafka_ex/gen_consumer.ex

@@ -209,6 +209,7 @@ defmodule KafkaEx.GenConsumer do
           | {:commit_threshold, non_neg_integer}
           | {:auto_offset_reset, :none | :earliest | :latest}
           | {:api_versions, map()}
+          | {:shutdown, timeout()}
           | {:extra_consumer_args, map()}
 
   @typedoc """
@@ -438,6 +439,8 @@ defmodule KafkaEx.GenConsumer do
   * `:fetch_options` - Optional keyword list that is passed along to the
     `KafkaEx.fetch` call.
 
+  * `:shutdown` - Optional amount of time in milliseconds that the supervisor will wait for a `GenConsumer` to terminate after emitting a `Process.exit(child, :shutdown)` signal. Defaults to `5_000`.
+
   * `:extra_consumer_args` - Optional parameter that is passed along to the
     `GenConsumer.init` call in the consumer module. Note that if `init/3` is not
     implemented, the default implementation calls to `init/2`, dropping the extra

lib/kafka_ex/gen_consumer/supervisor.ex

@@ -15,6 +15,8 @@ defmodule KafkaEx.GenConsumer.Supervisor do
 
   use DynamicSupervisor
 
+  @default_worker_shutdown 5_000
+
   @doc """
   Starts a `GenConsumer.Supervisor` process linked to the current process.
 
@@ -57,7 +59,8 @@ defmodule KafkaEx.GenConsumer.Supervisor do
         id: gen_consumer_module,
         start:
           {gen_consumer_module, :start_link,
-           [consumer_module, group_name, topic, partition, opts]}
+           [consumer_module, group_name, topic, partition, opts]},
+        shutdown: Keyword.get(opts, :shutdown, @default_worker_shutdown)
       }
     end