title: Spark Structured Streaming groupByKey on a time Window not working
body: <p>I need to batch my Kafka stream into time windows of 10 minutes each and then run some batch processing on it.</p>
  
  <p>Note: records below have a timestamp field</p>
  
  <pre><code>   val records = spark.readStream
        .format("kafka")
        .option("kafka.bootstrap.servers", brokerPool)
        .option("subscribe", topic)
        .option("startingOffsets", kafkaOffset)
        .load()
  </code></pre>
  
  <p>I add a time window to each record using,</p>
  
  <pre><code>.withColumn("window", window($"timing", windowDuration))
  </code></pre>
  
  <p>I created some helper classes like</p>
  
  <pre><code>  case class TimingWindow(
        start: java.sql.Timestamp,
        end: java.sql.Timestamp
    )
    case class RecordWithWindow(
        record: MyRecord,
        groupingWindow: TimingWindow
    )
  </code></pre>
  
  <p>Now I have a DF of type [RecordWithWindow]</p>
  
  <p>All this works very well.</p>
  
  <p>Next,</p>
  
  <pre><code>metricsWithWindow
    .groupByKey(_.groupingWindow)
    //By grouping, I get several records per time window
    //resulting an object of the below type which I write out to HDFS
  
    case class WindowWithRecords(
        records: Seq[MyRecord],
        window: TimingWindow
    )
  </code></pre>
  
  <p>Where I examine HDFS, </p>
  
  <p><strong>Example</strong>:</p>
  
  <p><em>Expected</em> : 
  Each WindowWithRecords object having a unique TimingWindow</p>
  
  <pre><code>WindowWithRecordsA(TimingWindowA, Seq(MyRecordA, MyRecordB, MyRecordC))
  </code></pre>
  
  <p><em>Actual</em> : 
  More than one WindowWithRecords object with the same TimingWindow</p>
  
  <pre><code>WindowWithRecordsA(TimingWindowA, Seq(MyRecordA, MyRecordB))
  WindowWithRecordsB(TimingWindowA, Seq(MyRecordC))
  </code></pre>
  
  <p>Looks like the groupByKey logic is not working well.</p>
  
  <p>I hope my question is clear. Any pointers would be helpful.</p>