Closes batchsender by sending the last batch immediately.

Author: marianogappa

scheduler/batchsender/batch_sender.go

@@ -69,6 +69,17 @@ func (bs *BatchSender) flush(items ...any) {
 
 	bs.items = append(bs.items, items...)
 
+	if len(bs.items) == 0 {
+		return
+	}
+
 	bs.sendFn(bs.items)
 	bs.items = nil
 }
+
+func (bs *BatchSender) Close() {
+	if bs.timer != nil {
+		bs.timer.Stop()
+	}
+	bs.flush()
+}

scheduler/scheduler_dfs.go

@@ -128,6 +128,7 @@ func (s *syncClient) resolveTableDfs(ctx context.Context, table *schema.Table, c
 	for r := range res {
 		batchSender.Send(r)
 	}
+	batchSender.Close()
 
 	// we don't need any waitgroups here because we are waiting for the channel to close
 	endTime := time.Now()