Abris #256 limit offset reset interval in ConfluentKafkaAvroReader ex…

…ample (cherry picked from commit de286a7)
AbsaOSS · Dec 21, 2021 · 0829165 · 0829165
1 parent 51993fa
commit 0829165
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/src/main/scala/za/co/absa/abris/examples/ConfluentKafkaAvroReader.scala b/src/main/scala/za/co/absa/abris/examples/ConfluentKafkaAvroReader.scala
@@ -18,10 +18,13 @@ package za.co.absa.abris.examples
 
 import org.apache.spark.sql.SparkSession
 import org.apache.spark.sql.functions.col
+import org.apache.spark.sql.streaming.Trigger
 import za.co.absa.abris.avro.read.confluent.SchemaManagerFactory
 import za.co.absa.abris.avro.registry.SchemaSubject
 import za.co.absa.abris.config.AbrisConfig
 
+import scala.concurrent.duration.DurationInt
+
 object ConfluentKafkaAvroReader {
 
   val kafkaTopicName = "test_topic"
@@ -63,6 +66,7 @@ object ConfluentKafkaAvroReader {
     deserialized
       .writeStream
       .format("console")
+      .trigger(Trigger.ProcessingTime(5.seconds))
       .option("truncate", "false")
       .start()
       .awaitTermination()