Exceptions are swallowed by KafkaProducer#sendAwait

[fabiankluempers]

Hi, there is a bug in the KafkaProducer#sendAwait.

It checks whether the returned RecordMetadata is null and resumes the continuation with the exception if that is the case.
But a non-null RecordMetadata is passed to the Callback in every case. It just has no offset (-1) and no partition (-1). This means all exceptions are swallowed.

An easy fix is to just check if the exception passed to the Callback is not null and to resume the continuation with that exception if that is the case.

[nomisRev]

Hey @fabiankluempers,

Thank you for opening this ticket, an alternative implementation has been written for producing elements that will be released soon. It is defined on top of Producer so it should fix this issue, but I will keep this open until I've released that in a next version.

You can check out the API, and implementation here.
Any feedback would be greatly appreciated, #156

[lsafer-meemer]

@nomisRev you might replace:

public suspend fun <A, B> KafkaProducer<A, B>.sendAwait(
  record: ProducerRecord<A, B>,
): RecordMetadata =
  suspendCoroutine { cont ->
    send(record) { a, e ->
      if (a != null) cont.resume(a) else cont.resumeWithException(e)
    }
  }

With:

public suspend fun <A, B> KafkaProducer<A, B>.sendAwait(
  record: ProducerRecord<A, B>,
): RecordMetadata =
  suspendCancellableCoroutine { cont ->
    send(record) { a, e ->
      if (a != null) cont.resume(a) else cont.resumeWithException(e)
    }
  }

It fixes the issue completely (I tried it myself)

[nomisRev]

Thank you @lsafer-meemer, but I have already solved this problem by building a much better solution.

Publishing now works like this, and doesn't swallow any exceptions. It also allows to offer instead of sendAndAwait which is much faster and recommended by Kafka.

I'm releasing in coming 2 weeks a new version of Kotlin Kafka, with these updates and updated documentation. Sorry for the wait. It's been busy during the end of last year, and beginning of this year.

publisher.publishScope {
  offer((1..10).map {
    ProducerRecord(topic.name(), "$it", "msg-$it")
  })
  publish((11..20).map {
    ProducerRecord(topic.name(), "$it", "msg-$it")
  })
  transaction {
    // transaction { } illegal to be called here DslMarker magic
    offer((21..30).map {
      ProducerRecord(topic.name(), "$it", "msg-$it")
    })
    publish((31..40).map {
      ProducerRecord(topic.name(), "$it", "msg-$it")
    })
  }// Waits until all offer finished in transaction, fails if any failed

  // looping
  (0..100).forEach {
    delay(100.milliseconds)
    val record = ProducerRecord(topic.name(), "$it", "msg-$it")
    offer(record)
  }
  
  // streaming
  flow(1..100)
    .onEach { delay(100.milliseconds) }
    .map { ProducerRecord(topic.name(), "$it", "msg-$it") }
    .collect { offer(it) }
}

[lsafer-meemer]

@nomisRev it actually looks better and well organized. cant wait for the new release 🔥

I know this is not the right place to ask this. but, please care about the ease of loading settings and configurations. Dummer people like me might choose to stick with the java APIs just because the better APIs needs configurations that they don't understand or configurations that are too implicit.

[nomisRev]

I know this is not the right place to ask this. but, please care about the ease of loading settings and configurations. Dummer people like me might choose to stick with the java APIs just because the better APIs needs configurations that they don't understand or configurations that are too implicit.

I'm not 100% what you mean @lsafer-meemer 🤔 Do you have any examples? My goal for the library is to stay as close to possible to the Java SDK, but offer better typed solutions and higher level APIs. Should be super simple to use, and not look completely foreign.

So in the above DSL ProducerRecord is just the Java SDK type.