Kafka 和 akka (scala)：如何创建 Source[CommittableMessage[Array[Byte], String], Consumer.Control]？答案

【问题标题】：Kafka and akka (scala): How to create Source[CommittableMessage[Array[Byte], String], Consumer.Control]?Kafka 和 akka (scala)：如何创建 Source[CommittableMessage[Array[Byte], String], Consumer.Control]？
【发布时间】：2018-05-18 23:25:37
【问题描述】：

我希望单元测试创建一个具有可提交消息和消费者控制的源。

或者转换这样创建的源：

val message: Source[Array[Byte], NotUsed] = Source.single("one message".getBytes)

这样的事情

Source[CommittableMessage[Array[Byte], String], Consumer.Control]

目标是对消息上的actor行为进行单元测试，而无需在构建机器上安装kafka

【问题讨论】：

标签： scala apache-kafka akka

【解决方案1】：

使用Consumer.committableSource 创建Source[CommittableMessage[K, V], Control]。这个想法是，在您的测试中，您将针对某个主题生成一条或多条消息，然后使用 committableSource 从同一主题中消费。

以下是说明此方法的示例：它是对 Akka Streams Kafka 项目中IntegrationSpec 的稍作调整的摘录。 IntegrationSpec 使用 scalatest-embedded-kafka，它为 ScalaTest 规范提供内存中的 Kafka 实例。

Source(1 to 100)
    .map(n => new ProducerRecord(topic1, partition0, null: Array[Byte], n.toString))
    .runWith(Producer.plainSink(producerSettings))

val consumerSettings = createConsumerSettings(group1)

val (control, probe1) = Consumer.committableSource(consumerSettings, TopicSubscription(Set(topic1)))
  .filterNot(_.record.value == InitialMsg)
  .mapAsync(10) { elem =>
    elem.committableOffset.commitScaladsl().map { _ => Done }
  }
  .toMat(TestSink.probe)(Keep.both)
  .run()

probe1
  .request(25)
  .expectNextN(25).toSet should be(Set(Done))

probe1.cancel()
Await.result(control.isShutdown, remainingOrDefault)

【讨论】：

【解决方案2】：

你可以使用这个助手来创建一个 CommittableMessage：

package akka.kafka.internal

import akka.Done
import akka.kafka.ConsumerMessage.{CommittableMessage, CommittableOffsetBatch, GroupTopicPartition, PartitionOffset}
import akka.kafka.internal.ConsumerStage.Committer
import org.apache.kafka.clients.consumer.ConsumerRecord

import scala.collection.immutable
import scala.concurrent.Future

object AkkaKafkaHelper {

  private val committer = new Committer {
    def commit(offsets: immutable.Seq[PartitionOffset]): Future[Done] = Future.successful(Done)
    def commit(batch: CommittableOffsetBatch): Future[Done] = Future.successful(Done)
  }

  def commitableMessage[K, V](key: K, value: V, topic: String = "topic", partition: Int = 0, offset: Int = 0, groupId: String = "group"): CommittableMessage[K, V] = {
    val partitionOffset = PartitionOffset(GroupTopicPartition(groupId, topic, partition), offset)
    val record = new ConsumerRecord(topic, partition, offset, key, value)
    CommittableMessage(record, ConsumerStage.CommittableOffsetImpl(partitionOffset)(committer))
  }
}

【讨论】：