Ensure threads are running when calling shutdown

Similar to the previous commit, if a worker thread unexpectedly dies
with unhandled messages in its queue and then we try to call `shutdown`,
those messages won't be handled unless `produce` is called in the future
(in which case the worker thread will read the `shutdown` message
enqueued here and then exit again). With this change we will restart the
worker thread so it can receive the `shutdown` message.

Author: cstyles

lib/kafka/async_producer.rb

@@ -144,6 +144,8 @@ def deliver_messages
     # @see Kafka::Producer#shutdown
     # @return [nil]
     def shutdown
+      ensure_threads_running!
+
       @timer_thread && @timer_thread.exit
       @queue << [:shutdown, nil]
       @worker_thread && @worker_thread.join