DDL notifier does not guarantee delivery when internal SQL failed to COMMIT

[lance6716]

Bug Report

tidb/pkg/ddl/notifier/subscribe.go

Lines 275 to 296 in c34a6b6

	defer func() {
	if err == nil {
	err = errors.Trace(session.Commit(ctx))
	} else {
	session.Rollback()
	}
	}()
	now := time.Now()
	if err = handler(ctx, session.Context, change.event); err != nil {
	return errors.Trace(err)
	}
	if time.Since(now) > slowHandlerLogThreshold {
	logutil.Logger(ctx).Warn("Slow process event",
	zap.Stringer("handler", handlerID),
	zap.Int64("ddlJobID", change.ddlJobID),
	zap.Int64("subJobID", change.subJobID),
	zap.Stringer("event", change.event),
	zap.Duration("duration", time.Since(now)))
	}
	newFlag := change.processedByFlag | (1 << handlerID)

If handler runs without error, DDL notifier marks the in-memory structure change as processed in line 295. However the handler may still fail in line 277 when commits. The change uses the same SQL transaction so the wrong processed state is not persisted, but the memory state of change is used afterward and causes problems

tidb/pkg/ddl/notifier/subscribe.go

Lines 233 to 244 in c34a6b6

	if change.processedByFlag == n.handlersBitMap {
	s3, err3 := n.sysSessionPool.Get()
	if err3 != nil {
	return errors.Trace(err3)
	}
	sess4Del := sess.NewSession(s3.(sessionctx.Context))
	err3 = n.store.DeleteAndCommit(
	ctx,
	sess4Del,
	change.ddlJobID,
	int(change.subJobID),
	)

1. Minimal reproduce step (Required)

Will add UT in fix PR

2. What did you expect to see? (Required)

3. What did you see instead (Required)

4. What is your TiDB version? (Required)