将Lagom连接到外部Kafka服务器(在Confluent Cloud上)

libraryDependencies ++= Seq(
  "com.typesafe.akka" %% "akka-stream-kafka" % "3.0.0",
  "com.lightbend.lagom" %% "lagom-scaladsl-kafka-client" % "1.6.4"
)

kafka-clients {
  bootstrap.servers = "your-kafka-server:9092"
  security.protocol = "SASL_SSL"
  sasl.mechanism = "PLAIN"
  sasl.jaas.config = "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"your-username\" password=\"your-password\";"
}

import akka.actor.ActorSystem
import akka.kafka.ProducerSettings
import akka.kafka.scaladsl.Producer
import akka.stream.scaladsl.Source
import org.apache.kafka.clients.producer.ProducerRecord
import org.apache.kafka.common.serialization.StringSerializer

object KafkaProducerExample {
  implicit val system = ActorSystem()

  val producerSettings = ProducerSettings(system, new StringSerializer, new StringSerializer)
    .withBootstrapServers("your-kafka-server:9092")
    .withProperty("security.protocol", "SASL_SSL")
    .withProperty("sasl.mechanism", "PLAIN")
    .withProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"your-username\" password=\"your-password\";")

  def sendMessage(topic: String, message: String): Unit = {
    Source.single(message)
      .map(value => new ProducerRecord[String, String](topic, value))
      .runWith(Producer.plainSink(producerSettings))
  }
}

import akka.actor.ActorSystem
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.scaladsl.Sink
import org.apache.kafka.common.serialization.StringDeserializer

object KafkaConsumerExample {
  implicit val system = ActorSystem()

  val consumerSettings = ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
    .withBootstrapServers("your-kafka-server:9092")
    .withGroupId("group1")
    .withProperty("security.protocol", "SASL_SSL")
    .withProperty("sasl.mechanism", "PLAIN")
    .withProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"your-username\" password=\"your-password\";")

  def consumeMessages(topic: String): Unit = {
    Consumer.plainSource(consumerSettings, Subscriptions.topics(topic))
      .map(record => record.value())
      .runWith(Sink.foreach(println))
  }
}