Add a Producer sink for producing record chunks (#35)

* Draft of Producer sink

* Remove unused import

* Remove passthrough functionality that was not going to work

* Remove inefficient non-chunked sink

src/main/scala/zio/kafka/client/Producer.scala

@@ -3,13 +3,13 @@ package zio.kafka.client
 import java.util.concurrent.Future
 import java.util.concurrent.atomic.AtomicLong
 
-import scala.collection.JavaConverters._
-
 import org.apache.kafka.clients.producer.{ Callback, KafkaProducer, ProducerRecord, RecordMetadata }
 import org.apache.kafka.common.serialization.Serde
-
-import zio.{ Chunk, Promise, UIO, ZIO, ZManaged }
+import zio._
 import zio.blocking._
+import zio.stream.ZSink
+
+import scala.collection.JavaConverters._
 
 trait Producer[K, V] {
   def produce(record: ProducerRecord[K, V]): BlockingTask[RecordMetadata]
@@ -96,4 +96,18 @@ object Producer {
       .map(unsafeMake)
   }
 
+  /**
+   * Sink that produces records to Kafka in chunks
+   *
+   * @param settings
+   * @tparam K
+   * @tparam V
+   * @return
+   */
+  def sink[K: Serde, V: Serde](
+    settings: ProducerSettings
+  ): ZManaged[Blocking, Throwable, ZSink[Blocking, Throwable, Nothing, Chunk[ProducerRecord[K, V]], Unit]] =
+    make[K, V](settings).map { producer =>
+      ZSink.drain.contramapM(producer.produceChunk)
+    }
 }

src/main/scala/zio/kafka/client/ProducerSettings.scala

@@ -6,7 +6,7 @@ import org.apache.kafka.clients.producer.ProducerConfig
 case class ProducerSettings(
   bootstrapServers: List[String],
   closeTimeout: Duration,
-  extraDriverSettings: Map[String, String]
+  extraDriverSettings: Map[String, AnyRef]
 ) {
   def driverSettings: Map[String, AnyRef] =
     Map(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG -> bootstrapServers.mkString(",")) ++