// PurgeRelay purges relay log files.
PurgeRelay purges relay log files.
// PurgeRelay purges relay log files .
func (w *SourceWorker) PurgeRelay(ctx context.Context, req *pb.PurgeRelayRequest) error {
	if w.closed.Load() {
		return terror.ErrWorkerAlreadyClosed.Generate()
	}
	if !w.relayEnabled.Load() {
		w.l.Warn("enable-relay is false, ignore purge relay")
		return nil
	}
	if !w.subTaskEnabled.Load() {
		w.l.Info("worker received purge-relay but didn't handling subtasks, read global checkpoint to decided active relay log")
		uuid := w.relayHolder.Status(nil).RelaySubDir
		_, _, subTaskCfgs, _, err := w.fetchSubTasksAndAdjust()
		if err != nil {
			return err
		}
		for _, subTaskCfg := range subTaskCfgs {
			loc, err2 := getMinLocForSubTaskFunc(ctx, subTaskCfg)
			if err2 != nil {
				return err2
			}
			w.l.Info("update active relay log with",
				zap.String("task name", subTaskCfg.Name),
				zap.String("uuid", uuid),
				zap.String("binlog name", loc.Position.Name))
			if err3 := streamer.GetReaderHub().UpdateActiveRelayLog(subTaskCfg.Name, uuid, loc.Position.Name); err3 != nil {
				w.l.Error("Error when update active relay log", zap.Error(err3))
			}
		}
	}
	return w.relayPurger.Do(ctx, req)
}
func ( w * SourceWorker ) PurgeRelay ( ctx context . Context , req * pb . PurgeRelayRequest ) error { if w . closed . Load ( ) { return terror . ErrWorkerAlreadyClosed . Generate ( ) } if ! w . relayEnabled . Load ( ) { w . l . Warn ( " " ) return nil } if ! w . subTaskEnabled . Load ( ) { w . l . Info ( " " ) uuid := w . relayHolder . Status ( nil ) . RelaySubDir _ , _ , subTaskCfgs , _ , err := w . fetchSubTasksAndAdjust ( ) if err != nil { return err } for _ , subTaskCfg := range subTaskCfgs { loc , err2 := getMinLocForSubTaskFunc ( ctx , subTaskCfg ) if err2 != nil { return err2 } w . l . Info ( " " , zap . String ( " " , subTaskCfg . Name ) , zap . String ( " " , uuid ) , zap . String ( " " , loc . Position . Name ) ) if err3 := streamer . GetReaderHub ( ) . UpdateActiveRelayLog ( subTaskCfg . Name , uuid , loc . Position . Name ) ; err3 != nil { w . l . Error ( " " , zap . Error ( err3 ) ) } } } return w . relayPurger . Do ( ctx , req ) }
PurgeRelay purges relay log files.
PurgeRelay purges relay log files .