fix error

cdc/sink/dmlsink/cloudstorage/dml_worker.go

@@ -19,14 +19,14 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/metrics"
 	mcloudstorage "github.com/pingcap/tiflow/cdc/sink/metrics/cloudstorage"
 	"github.com/pingcap/tiflow/engine/pkg/clock"
 	"github.com/pingcap/tiflow/pkg/chann"
+	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/sink/cloudstorage"
 	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	"github.com/prometheus/client_golang/prometheus"
@@ -248,13 +248,20 @@ func (d *dmlWorker) writeDataFile(ctx context.Context, path string, task *single
 		if inErr != nil {
 			return 0, inErr
 		}
+
 		defer func() {
-			if inErr == nil {
-				inErr = writer.Close(ctx)
+			closeErr := writer.Close(ctx)
+			if inErr != nil {
+				log.Error("failed to close writer", zap.Error(closeErr),
+					zap.Int("workerID", d.id),
+					zap.Any("table", task.tableInfo.TableName),
+					zap.String("namespace", d.changeFeedID.Namespace),
+					zap.String("changefeed", d.changeFeedID.ID))
+				if inErr == nil {
+					inErr = closeErr
+				}
 			}
-			// TODO: maybe we should abort the MultipartUpload here.
 		}()
-
 		if _, inErr = writer.Write(ctx, buf.Bytes()); inErr != nil {
 			return 0, inErr
 		}