feat: added new integration points with java kafka producer and consu…

…mer that rely on interface instead of implementation

zio-kafka/src/main/scala/zio/kafka/consumer/Consumer.scala

@@ -1,6 +1,12 @@
 package zio.kafka.consumer
 
-import org.apache.kafka.clients.consumer.{ ConsumerRecord, KafkaConsumer, OffsetAndMetadata, OffsetAndTimestamp }
+import org.apache.kafka.clients.consumer.{
+  Consumer => JConsumer,
+  ConsumerRecord,
+  KafkaConsumer,
+  OffsetAndMetadata,
+  OffsetAndTimestamp
+}
 import org.apache.kafka.common._
 import zio._
 import zio.kafka.consumer.diagnostics.DiagnosticEvent.Finalization
@@ -197,6 +203,22 @@ object Consumer {
       runloopAccess  <- RunloopAccess.make(settings, consumerAccess, diagnostics)
     } yield new ConsumerLive(consumerAccess, runloopAccess)
 
+  /**
+   * Create a zio-kafka Consumer from an org.apache.kafka Consumer
+   *
+   * You are responsible for creating and closing the Consumer
+   */
+  def fromJavaConsumer(
+    javaConsumer: JConsumer[Array[Byte], Array[Byte]],
+    settings: ConsumerSettings,
+    diagnostics: Diagnostics = Diagnostics.NoOp
+  ): ZIO[Scope, Throwable, Consumer] =
+    for {
+      _              <- ZIO.addFinalizer(diagnostics.emit(Finalization.ConsumerFinalized))
+      consumerAccess <- ConsumerAccess.make(javaConsumer)
+      runloopAccess  <- RunloopAccess.make(settings, consumerAccess, diagnostics)
+    } yield new ConsumerLive(consumerAccess, runloopAccess)
+
   /**
    * Accessor method
    */

zio-kafka/src/main/scala/zio/kafka/producer/Producer.scala

@@ -225,6 +225,25 @@ object Producer {
       _ <- ZIO.blocking(producer.sendFromQueue).forkScoped
     } yield producer
 
+  /**
+   * Create a zio-kafka Producer from an existing org.apache.kafka Producer
+   *
+   * You are responsible for creating and closing the Producer
+   */
+  def fromJavaProducer(
+    javaProducer: JProducer[Array[Byte], Array[Byte]],
+    settings: ProducerSettings
+  ): ZIO[Scope, Throwable, Producer] =
+    for {
+      runtime <- ZIO.runtime[Any]
+      sendQueue <-
+        Queue.bounded[(Chunk[ByteRecord], Promise[Nothing, Chunk[Either[Throwable, RecordMetadata]]])](
+          settings.sendBufferSize
+        )
+      producer = new ProducerLive(javaProducer, runtime, sendQueue)
+      _ <- ZIO.blocking(producer.sendFromQueue).forkScoped
+    } yield producer
+
   /**
    * Accessor method
    */