title: Spark streaming append output mode with trigger once
body: <p>I try to use Spark streaming (2.2.0) with Kafka input and such configuration:</p>
  
  <ul>
  <li>Watermarking and aggregation</li>
  <li>Windowing</li>
  <li>Output mode: OutputMode.Append()</li>
  <li>Trigger mode: Trigger.Once()</li>
  </ul>
  
  <p>But I have no output (neither in kafka nor in console).
  Launching the application multiple times does not work either.</p>
  
  <p>When I use OutputMode.Complete() or Trigger.ProcessingTime(0L), it works. But unfortunately, this is not my need.</p>
  
  <p>Is <strong>append</strong> mode supported with trigger <strong>once</strong> ?</p>
  
  <p>Here is a minimal application to reproduce the issue.
  <a href="https://gist.github.com/mprevel/84b86e6db3488e9045fbdcb61508a51e" rel="nofollow noreferrer">Gist</a></p>
  
  <pre class="lang-scala prettyprint-override"><code>case class Model(valueForGroupBy: Int, time: Timestamp)
  
  object Application {
  
    val appName = "sample"
  
  //  val outputMode: OutputMode = OutputMode.Complete() // OK
    val outputMode: OutputMode = OutputMode.Append() // KO with trigger once
  
    val triggerMode: Trigger = Trigger.Once()
  
    val delayThreshold: FiniteDuration = 1.minute // watermarking wait for late
  
    val duration : FiniteDuration = 1.minute // window duration and slide
  
    val topic = "SAMPLE_TOPIC"
    val bootstrapServers = "127.0.0.1:9092"
  
  
    type KafkaKV = (String, String)
  
    def main(args: Array[String]): Unit = {
  
      val spark: SparkSession = SparkSession
        .builder
        .appName(appName)
        .getOrCreate()
  
      import org.apache.spark.sql._
      import org.apache.spark.sql.functions._
      import spark.implicits._
  
      val streamReader: DataStreamReader = spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", bootstrapServers)
        .option("subscribe", topic)
        .option("startingOffsets", "earliest")
        .option("failOnDataLoss", "false")
  
      val df: DataFrame = streamReader.load()
  
      val ds: Dataset[Model] = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)").as[KafkaKV]
        .select(from_json($"value", Encoders.product[Model].schema).as("json"))
        .select($"json.*")
        .as[Model]
  
      val groupByColumns = Seq(
        window(new Column("time"), windowDuration = duration.toString, slideDuration = duration.toString),
        new Column("valueForGroupBy")
      )
  
      val agg = ds
        .withWatermark("time", delayThreshold.toString)
        .groupBy(groupByColumns: _*)
        .count()
  
      val streamWriter = agg
        .selectExpr(s"CAST(valueForGroupBy AS STRING) AS key", "to_json(struct(*)) AS value")
        .writeStream
        .trigger(triggerMode)
        .outputMode(outputMode)
        .format("console")
        .option("truncate", value = false)
  
      val streamingQuery = streamWriter.start()
  
      streamingQuery.awaitTermination()
  
    }
  }
  </code></pre>