卡夫卡和阿卡卡(scala)：如何创建源[CommittableMessage[Array[Byte]，String]，Consumer.Control]？

import akka.kafka.scaladsl.Consumer
import akka.kafka.scaladsl.Consumer.Control
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.{ByteArrayDeserializer, StringDeserializer}
import akka.kafka.{ConsumerSettings, Subscriptions}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import akka.kafka.scaladsl.Consumer.DrainingControl
import akka.kafka.ConsumerMessage.{CommittableMessage, CommittableOffsetBatch}

val consumerSettings = ConsumerSettings(system, new ByteArrayDeserializer, new StringDeserializer)
  .withBootstrapServers("kafka-broker1:9092,kafka-broker2:9092")
  .withGroupId("my-group")
  .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

val control = Consumer
  .committableSource(consumerSettings, Subscriptions.topics("my-topic"))
  .map { msg: CommittableMessage[Array[Byte], String] =>
    // 在这里处理消息
    // msg.record.value() 获取消息的值
    // msg.committableOffset 提供了提交偏移量的方法
    // ...
    msg
  }
  .toMat(Sink.ignore)(DrainingControl.apply)
  .run()