How to convert the following piece of java to kotlin?

consumer.commitAsync(new OffsetCommitCallback() {
    public void onComplete(Map<TopicPartition,
    OffsetAndMetadata> offsets, Exception e) {
        if (e != null)
            log.error("Commit failed for offsets {}", offsets, e);
    }
});

I got the above piece of codes from definition to kafka. I have no idea how to convert them into kotlin. I tried, but failed.

Any comments welcomed. Thanks

UPDATE

I converted to

object:OffsetCommitCallback() {
  override fun onComplete(offsets:Map<TopicPartition, OffsetAndMetadata>, e:Exception) {
    if (e != null) log.error("Commit failed for offsets {}", offsets, e)
  }
}

but got This class does not have a constructor.

UPDATE

it seems that the following work:

kafkaConsumer.commitAsync(mapOf(k to currentOffsets.get(k)), object:OffsetCommitCallback {
  override fun onComplete(offsets:Map<TopicPartition, OffsetAndMetadata>, e:Exception) {
    e?.let {
      log.error("Commit failed for offsets {}", offsets, e)
    }
  }
})

It looks like OffsetCommitCallback is a functional interface, so you should be able to use a lambda in your function call (SAM conversion):

kafkaConsumer.commitAsync(mapOf(k to currentOffsets[k])) { offsets, e ->
    e?.let {
      log.error("Commit failed for offsets {}", offsets, e)
    }
}

It looks like OffsetCommitCallback is an interface, not a class, so creating an anonymous instance of it would not use parentheses:

object : OffsetCommitCallback { ... }

rather than

object : OffsetCommitCallback() { ... }

Note also that, given that the original code is checking whether or not e is null, you probably need to use a nullable type in your Kotlin method signature:

override fun onComplete(..., e: Exception?)