fix CDC invalid command batch cleanup (#26000)

## What type of PR is this?

- [ ] API-change
- [x] BUG
- [ ] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #25934

## What this PR does / why we need it:

fix CDC invalid command batch cleanup

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

Author: jiangxinmeng1

pkg/cdc/sinker_v2.go

@@ -353,6 +353,7 @@ func (s *mysqlSinker2) processCommand(ctx context.Context, cmd *Command) {
 			zap.String("table", s.dbTblInfo.String()),
 			zap.String("command", cmd.String()),
 			zap.Error(err))
+		cmd.Close()
 		s.SetError(err)
 		return
 	}

pkg/cdc/sinker_v2_test.go

@@ -303,6 +303,29 @@ func TestMysqlSinker2_CommandProcessing(t *testing.T) {
 
 		sinker.ClearError()
 	})
+
+	t.Run("InvalidBatchCommand_CleansUp", func(t *testing.T) {
+		mp, err := mpool.NewMPool("invalid-command-cleanup", 0, mpool.NoFixed)
+		require.NoError(t, err)
+		defer mpool.DeleteMPool(mp)
+
+		cmd := NewInsertBatchCommand(
+			&batch.Batch{Vecs: []*vector.Vector{vector.NewVec(types.T_int32.ToType())}},
+			mp,
+			types.BuildTS(100, 0),
+			types.BuildTS(200, 0),
+		)
+		require.NotNil(t, cmd.Mp)
+
+		sinker.processCommand(ctx, cmd)
+
+		require.Error(t, sinker.Error())
+		assert.Contains(t, sinker.Error().Error(), "InsertBatch has no rows")
+		assert.Nil(t, cmd.InsertBatch)
+		assert.Nil(t, cmd.Mp)
+
+		sinker.ClearError()
+	})
 }
 
 func TestMysqlSinker2_Reset(t *testing.T) {