Kafkaトピックからストリームを読み取るときに、Spark Structured Streamingにタイムアウトの問題がありますか？

構造化ストリーミングでforeachbatchを使用してkafkaトピックからストリームを読み取るSparkジョブを実装しました。

val df = spark.readStream
  .format("kafka")
  .option("kafka.bootstrap.servers", "mykafka.broker.io:6667")
  .option("subscribe", "test-topic")
  .option("kafka.security.protocol", "SASL_SSL")
  .option("kafka.ssl.truststore.location", "/home/hadoop/cacerts")
  .option("kafka.ssl.truststore.password", tspass)
  .option("kafka.ssl.truststore.type", "JKS")
  .option("kafka.sasl.kerberos.service.name", "kafka")
  .option("kafka.sasl.mechanism", "GSSAPI")
  .option("groupIdPrefix","MY_GROUP_ID")
  .load()

val streamservice = df.selectExpr("CAST(value AS STRING)")
  .select(from_json(col("value"), schema).as("data"))
  .select("data.*")


var stream_df = streamservice
  .selectExpr("cast(id as string) id", "cast(x as int) x")

val monitoring_stream = stream_df.writeStream
  .trigger(Trigger.ProcessingTime("120 seconds"))
  .foreachBatch { (batchDF: DataFrame, batchId: Long) =>
    if(!batchDF.isEmpty) { }
  }
  .start()
  .awaitTermination()

以下の質問があります。

1. kafkaトピックに長期間データがない場合、stream_df.writeStreamは自動的に終了しますか？これにタイムアウト制御はありますか？

2. kafkaトピックがkafkaブローカーから削除された場合、stream_df.writeStreamは終了しますか？

上記の2つのケースで、sparkジョブが終了せずにkafkaトピックを監視し続けることを願っています。kafkaコネクタやstream_df.writerstreamに特別な設定が必要ですか？