backup: handle final flush errors when compacting backups

When processing spans for compaction, we always perform a final flush of
the SSTKeyWriter via `defer`. However, we currently do not propagate
that error anywhere and it is swallowed after it is logged. This updates
the compaction processor to properly handle `Flush` errors.

Epic: None

Release note: None

Author: kev-cao

pkg/backup/compaction_processor.go

@@ -240,7 +240,7 @@ func (p *compactBackupsProcessor) processSpanEntries(
 	entryCh chan execinfrapb.RestoreSpanEntry,
 	encryption *jobspb.BackupEncryptionOptions,
 	store cloud.ExternalStorage,
-) error {
+) (err error) {
 	var fileEncryption *kvpb.FileEncryptionOptions
 	if encryption != nil {
 		fileEncryption = &kvpb.FileEncryptionOptions{Key: encryption.Key}
@@ -256,12 +256,7 @@ func (p *compactBackupsProcessor) processSpanEntries(
 	if err != nil {
 		return err
 	}
-	defer func() {
-		if err := sink.Flush(ctx); err != nil {
-			log.Warningf(ctx, "failed to flush sink: %v", err)
-			logClose(ctx, sink, "SST sink")
-		}
-	}()
+	defer logClose(ctx, sink, "failed to close sst sink")
 	pacer := newBackupPacer(
 		ctx, p.FlowCtx.Cfg.AdmissionPacerFactory, p.FlowCtx.Cfg.Settings,
 	)
@@ -274,7 +269,7 @@ func (p *compactBackupsProcessor) processSpanEntries(
 			return ctx.Err()
 		case entry, ok := <-entryCh:
 			if !ok {
-				return nil
+				return sink.Flush(ctx)
 			}
 			if assigned, err := p.isAssignedEntry(entry); err != nil {
 				return err