Estou tentando agregar a contagem de registros a cada 10 segundos usando o streaming estruturado para os seguintes dados Kafka recebidos

{ 
  "ts2" : "2018/05/01 00:02:50.041", 
  "serviceGroupId" : "123", 
  "userId" : "avv-0", 
  "stream" : "", 
  "lastUserActivity" : "00:02:50", 
  "lastUserActivityCount" : "0" 
} 
{ 
  "ts2" : "2018/05/01 00:09:02.079", 
  "serviceGroupId" : "123", 
  "userId" : "avv-0", 
  "stream" : "", 
  "lastUserActivity" : "00:09:02", 
  "lastUserActivityCount" : "0" 
} 
{ 
  "ts2" : "2018/05/01 00:09:02.086", 
  "serviceGroupId" : "123", 
  "userId" : "avv-2", 
  "stream" : "", 
  "lastUserActivity" : "00:09:02", 
  "lastUserActivityCount" : "0" 
} 

Com a seguinte lógica

val sdvTuneInsAgg1 = df 
  .withWatermark("ts2", "10 seconds") 
  .groupBy(window(col("ts2"),"10 seconds")) 
  .agg(count("*") as "count") 
  .as[CountMetric1] 

val query1 = sdvTuneInsAgg1.writeStream
      .format("console")
      .foreach(writer)
      .start()

e não vejo nenhum registro dentro do escritor. Mas, a única anomalia é que a data atual é 24/05/2018, mas o registro que estou processando (ts2) tem datas antigas. A agregação / contagem funcionará neste cenário?