在这里，我建议使用可变列表轻松解决您的问题：

val kafkaMessages: Source[ConsumerMessage.CommittableMessage[String, Array[Byte]], Consumer.Control] = Consumer.committableSource(
    consumerSettings,
    Subscriptions.topics(conf.getString("kafka.topics.errorRawCdr")))

    import scala.collection.mutable._
    val errorMessages: MutableList[String] = new MutableList

  val cdrs: Source[Errors, Consumer.Control] = kafkaMessages.map(msg => {
    val bytes: Array[Byte] = msg.record.value()
    val errormsg = (bytes.map(_.toChar)).mkString
    errorMessages += errormsg
    new Errors(1235, "filename", "cdr", "cdr_type", 0, errormsg)
  })

  cdrs.to(new ErrorKuduSink(session, table)).run()

解决方案1：在Map方法中发送电子邮件（将在每个Kafka消息中发送电子邮件）

def sendEmail(errormsg: String): Unit = ???

val cdrs: Source[Errors, Consumer.Control] = 
  kafkaMessages.map { msg => 
    val bytes: Array[Byte] = msg.record.value()
    val errormsg = (bytes.map(_.toChar)).mkString
    sendEmail(errormsg) // call function that sends email
    new Errors(1235, "filename", "cdr", "cdr_type", 0, errormsg)
  }

解决方案2：如果你想更复杂地使用 errormsg 在下游阶段，您需要从Map阶段返回一个元组：

val kafkaMessages: Source[ConsumerMessage.CommittableMessage[String, Array[Byte]], Consumer.Control] = 
  Consumer.committableSource(consumerSettings, Subscriptions.topics(conf.getString("kafka.topics.errorRawCdr")))

val cdrs: Source[Errors, Consumer.Control] = 
  kafkaMessages.map { msg => 
    val bytes: Array[Byte] = msg.record.value()
    val errormsg = (bytes.map(_.toChar)).mkString
    (new Errors(1235, "filename", "cdr", "cdr_type", 0, errormsg), errormsg) // we are returning a tuple so type of downstream elements will be (Errors, String)
  }.map { case i@(errors, errormsg) => 
    sendEmail(errormsg)
    i
  }.map { tuple =>
    ...
  }.map(_._1) // as we dont need a tuple any more we can get original element and continue processing of it

cdrs.to(new ErrorKuduSink(session, table)).run()

解决方案3：如果您想要更复杂的处理（例如在一封电子邮件中批处理多个errormsg），您可能需要创建一个runnablegraph

val g = RunnableGraph.fromGraph(GraphDSL.create() { implicit builder: GraphDSL.Builder[NotUsed] =>
  import GraphDSL.Implicits._
  val in = Consumer.committableSource(consumerSettings, Subscriptions.topics(conf.getString("kafka.topics.errorRawCdr")))
    .map { msg => 
      val bytes: Array[Byte] = msg.record.value()
      val errormsg = (bytes.map(_.toChar)).mkString
      (new Errors(1235, "filename", "cdr", "cdr_type", 0, errormsg), errormsg)
    }
  val kuduout = new ErrorKuduSink(session, table)
  val emailout = Sink.foreach[Seq[String]] { errormsgs =>
    sendEmail(errormsgs)
  }
  val f1 = Flow[(Errors, String)]
    .map(_._1) // take errors

  val f2 = Flow[(Errors, String)]
    .map(_._2) // take errormsgs
    .groupedWithin(100, 1.hour)

  val bcast = builder.add(Broadcast[Int](2))

  in ~> bcast
  bcast ~> f1 ~> kuduout
  bcast ~> f2 ~> emailout 
  ClosedShape
})