如何在启用X-Pack的情况下设置Elasticsearch Structured Streaming？

val spark = SparkSession.builder()
  .appName("Elasticsearch Structured Streaming")
  .master("local[*]")
  .config("es.nodes", "localhost")
  .config("es.port", "9200")
  .config("es.net.http.auth.user", "username")
  .config("es.net.http.auth.pass", "password")
  .getOrCreate()

import spark.implicits._

val streamingDF = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "localhost:9092")
  .option("subscribe", "topic")
  .load()
  .selectExpr("CAST(value AS STRING)")
  .as[String]

streamingDF.writeStream
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    batchDF.write
      .format("es")
      .option("es.resource", "index/type")
      .option("es.nodes", "localhost")
      .option("es.port", "9200")
      .option("es.net.http.auth.user", "username")
      .option("es.net.http.auth.pass", "password")
      .mode("append")
      .save()
  }
  .start()
  .awaitTermination()