make ThreadedProducer clonable like BaseProducer

The threaded producer was not `Clone` due to handling the background
thread join handle but that can easily be stored in an `Arc` that the
last drop of the client unwraps and terminates the background thread.

`Arc::into_inner` guarantees that exactly one thread will manage to
unwrap the handle.

Author: petrosagg

src/producer/base_producer.rs

@@ -639,13 +639,14 @@ where
 /// queued events, such as delivery notifications. The thread will be
 /// automatically stopped when the producer is dropped.
 #[must_use = "The threaded producer will stop immediately if unused"]
+#[derive(Clone)]
 pub struct ThreadedProducer<C, Part: Partitioner = NoCustomPartitioner>
 where
     C: ProducerContext<Part> + 'static,
 {
     producer: Arc<BaseProducer<C, Part>>,
     should_stop: Arc<AtomicBool>,
-    handle: Option<JoinHandle<()>>,
+    handle: Option<Arc<JoinHandle<()>>>,
 }
 
 impl FromClientConfig for ThreadedProducer<DefaultProducerContext, NoCustomPartitioner> {
@@ -687,7 +688,7 @@ where
         Ok(ThreadedProducer {
             producer,
             should_stop,
-            handle: Some(thread),
+            handle: Some(Arc::new(thread)),
         })
     }
 }
@@ -778,7 +779,7 @@ where
 {
     fn drop(&mut self) {
         trace!("Destroy ThreadedProducer");
-        if let Some(handle) = self.handle.take() {
+        if let Some(handle) = self.handle.take().and_then(Arc::into_inner) {
             trace!("Stopping polling");
             self.should_stop.store(true, Ordering::Relaxed);
             trace!("Waiting for polling thread termination");