Scala读取卡夫卡主题并写入MySQL表

import akka.actor.ActorSystem
import akka.kafka.scaladsl.Consumer
import akka.kafka.{ConsumerSettings, Subscriptions}
import akka.stream.scaladsl.Sink
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer
import java.sql.{Connection, DriverManager, PreparedStatement}

object KafkaToMySQL {
  def main(args: Array[String]): Unit = {
    implicit val system = ActorSystem()

    val bootstrapServers = "localhost:9092"
    val topic = "test-topic"
    val groupId = "group1"

    val consumerSettings = ConsumerSettings(system, new StringDeserializer, new StringDeserializer)
      .withBootstrapServers(bootstrapServers)
      .withGroupId(groupId)
      .withProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")

    Consumer.plainSource(consumerSettings, Subscriptions.topics(topic))
      .map(record => record.value())
      .runWith(Sink.foreach { value =>
        writeToMySQL(value)
      })
  }

  def writeToMySQL(data: String): Unit = {
    val url = "jdbc:mysql://localhost:3306/mydatabase"
    val username = "user"
    val password = "password"

    try (Connection conn = DriverManager.getConnection(url, username, password)) {
      val sql = "INSERT INTO mytable (data) VALUES (?)"
      val pstmt = conn.prepareStatement(sql)
      pstmt.setString(1, data)
      pstmt.executeUpdate()
    } catch {
      case e: Exception => e.printStackTrace()
    }
  }
}