fix: Flush DeleteRecord messages when batch writer is flushed (#2154)

The current logic for `BatchWriter` does not flush `DeleteRecord` messages when the writer is flushed. This PR adds that flushing.

I'm also wondering whether we should add a call to `flushDeleteRecordTables` in https://github.com/cloudquery/plugin-sdk/blob/main/writers/batchwriter/batchwriter.go#L307-L315. WDYT?

Author: maaarcelino

writers/batchwriter/batchwriter.go

@@ -107,7 +107,10 @@ func (w *BatchWriter) Flush(ctx context.Context) error {
 	if err := w.flushMigrateTables(ctx); err != nil {
 		return err
 	}
-	return w.flushDeleteStaleTables(ctx)
+	if err := w.flushDeleteStaleTables(ctx); err != nil {
+		return err
+	}
+	return w.flushDeleteRecordTables(ctx)
 }
 
 func (w *BatchWriter) Close(context.Context) error {
@@ -310,6 +313,9 @@ func (w *BatchWriter) Write(ctx context.Context, msgs <-chan message.WriteMessag
 			if err := w.flushDeleteStaleTables(ctx); err != nil {
 				return err
 			}
+			if err := w.flushDeleteRecordTables(ctx); err != nil {
+				return err
+			}
 			if err := w.startWorker(ctx, m); err != nil {
 				return err
 			}