monix/monix-kafka

Manual commitAsync completes before actual commit

开放

#94 创建于 2019年8月1日

 (3 条评论) (1 个反应) (0 位负责人)Scala (36 个派生)auto 404
buggood first issue

仓库指标

星标
 (124 个星标)
PR 合并指标
 (30 天内没有已合并 PR)

描述

In KafkaCOnsumerObservableManualCommit.scala:56 there's incorrect implementation of async commit

override def commitBatchAsync(batch: Map[TopicPartition, Long], callback: OffsetCommitCallback): Task[Unit] =
  Task {
    blocking(consumer.synchronized(consumer.commitAsync(batch.map {
      case (k, v) => k -> new OffsetAndMetadata(v)
     }.asJava, callback)))
  }

Apache kafka commitAsync(offsets, callback) returns immediately and invokes callback when commit completes

this task completes when consumer.commitAsync returns, not when callback invoked and also ignores possible commit errors also passed through callback, it should probably be rewritten to Task.async

贡献者指南