title: Make na abstraction to kafka consumer using ktor and kotlin
body: <p>I'm creating a abstraction for consumer and producer Kafka, to avoid duplicate code all the time. So i created a lib using kotlin and gradle, named "kafka-commons", and put the following code:
  For Kafka producer: </p>
  
  <pre><code>fun producer(
      bootstrapServers: String,
      idempotence: Boolean,
      acks: Acks,
      retries: Int,
      requestPerConnection: Int,
      compression: Compression,
      linger: Int,
      batchSize: BatchSize
  ): KafkaProducer&lt;String, Any&gt; {
      val prop: HashMap&lt;String, Any&gt; = HashMap()
      prop[BOOTSTRAP_SERVERS_CONFIG] = bootstrapServers
      prop[KEY_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java.name
      prop[VALUE_SERIALIZER_CLASS_CONFIG] = StringSerializer::class.java.name
      prop[ENABLE_IDEMPOTENCE_CONFIG] = idempotence
      prop[ACKS_CONFIG] = acks.value
      prop[RETRIES_CONFIG] = retries
      prop[MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION] = requestPerConnection
      prop[COMPRESSION_TYPE_CONFIG] = compression.value
      prop[LINGER_MS_CONFIG] = linger
      prop[BATCH_SIZE_CONFIG] = batchSize.value
  
      return KafkaProducer(prop)
  }
  
  suspend inline fun &lt;reified K : Any, reified V : Any&gt; KafkaProducer&lt;K, V&gt;.dispatch(record: ProducerRecord&lt;K, V&gt;) =
      suspendCoroutine&lt;RecordMetadata&gt; { continuation -&gt;
          val callback = Callback { metadata, exception -&gt;
              if (metadata == null) {
                  continuation.resumeWithException(exception!!)
              } else {
                  continuation.resume(metadata)
              }
          }
          this.send(record, callback)
      }
  </code></pre>
  
  <p>So, i created a default Command object with the following structure</p>
  
  <pre><code>data class Command(
      val id: UUID,
      val status: CommandStatus,
      val message: Any
  )
  </code></pre>
  
  <p>id: create a unique ID
  status: crete a message status (can be: Open /Processing / Closed / Error)
  message: an object from http reques (for example: If have an Post</p>
  
  <p>For example: if have a "insert user: POST" with body :</p>
  
  <blockquote>
    <p>{   "id": 1,   "name" : "John",   "lastName" : "Wick" }</p>
  </blockquote>
  
  <p>so message will be this object, and so on.</p>
  
  <p>And for create this command, i made this function:</p>
  
  <pre><code>suspend fun creatCommand(
      topicName: String,
      id: UUID,
      commandStatus: CommandStatus,
      request: Any,
      bootstrapServers: String,
      idempotence: Boolean,
      acks: Acks,
      retries: Int,
      requestPerConnection: Int,
      compression: Compression,
      linger: Int,
      batchSize: BatchSize
  ): Unit {
      val producer = producer(
          bootstrapServers,
          idempotence,
          acks,
          retries,
          requestPerConnection,
          compression,
          linger,
          batchSize)
  
      val command = toCommand(processStarted(id, commandStatus, request))
      val record = ProducerRecord&lt;String, Any&gt;(topicName, id.toString(), command)
      coroutineScope { launch { producer.dispatch(record) } }
  }
  </code></pre>
  
  <p>So, I created other APIs and just call this function to create a producer that sends a command to kafka. Like: </p>
  
  <pre><code>fun Route.user(service: Service) =
      route("/api/access") {
          post("/test") {
              call.respond(service.command(call.receive())) 
              }
  }
  
  &gt;&gt;&gt;&gt;&gt;&gt; other class &lt;&lt;&lt;&lt;&lt;&lt;&lt;&lt;
  
  classService () {
      fun command( all parameters) { creatCommand(all parameters)} 
  }
  </code></pre>
  
  <p>Sor far so good. All works great. </p>
  
  <p>Now my problemas begin. I'm trying to create a consumer. 
  First i made this: </p>
  
  <pre><code>fun consumer(
      bootstrapServers: String,
      group: String,
      autoCommit: Boolean,
      offsetBehaviour: OffsetBehaviour,
      pollMax: Int
  ): KafkaConsumer&lt;String, Any&gt; {
      val prop: HashMap&lt;String, Any&gt; = HashMap()
      prop[BOOTSTRAP_SERVERS_CONFIG] = bootstrapServers
      prop[KEY_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java.name
      prop[VALUE_DESERIALIZER_CLASS_CONFIG] = StringDeserializer::class.java.name
      prop[GROUP_ID_CONFIG] = group
      prop[AUTO_OFFSET_RESET_CONFIG] = offsetBehaviour
      prop[ENABLE_AUTO_COMMIT_CONFIG] = autoCommit
      prop[MAX_POLL_RECORDS_CONFIG] = pollMax
  
      return KafkaConsumer(prop)
  }
  </code></pre>
  
  <p>And after:</p>
  
  <pre><code>fun&lt;T&gt; recordingCommand(
      command: Class&lt;T&gt;,
      topic: String,
      bootstrapServers: String,
      group: String,
      autoCommit: Boolean,
      offsetBehaviour: OffsetBehaviour,
      pollMax: Int
  ) {
      val consumer = consumer(bootstrapServers, group, autoCommit, offsetBehaviour, pollMax)
      consumer.subscribe(mutableListOf(topic))
      while (true) {
          val records = consumer.poll(Duration.ofMillis(100))
          for (record in records) {
              val om = ObjectMaper
              om.readvalue(record.value(), command::class.java)
  
              &gt;&gt;&gt;&gt; I GOT LOST HERE &lt;&lt;&lt;&lt;&lt;&lt;
  
          }
      }
  }
  </code></pre>
  
  <p>What i need: creating a abstract consumer that record all data inside a Command.message() (only the message) in a database.</p>
  
  <p>For example, i need to record the user above (id 1, john wick ) into a postgresql database.
  So if i had a service with insert method, i can call it, passing insertmethod like:</p>
  
  <pre><code>service.insert(recordingCommand(all parameters)).
  </code></pre>
  
  <p>Anyone can help me with that?</p>