Method DropFlowSource

flow/activities/flowable.go:821–858

(ctx context.Context, req *protos.DropFlowActivityInput)

}

func (a *FlowableActivity) DropFlowSource(ctx context.Context, req *protos.DropFlowActivityInput) error {
	ctx = context.WithValue(ctx, shared.FlowNameKey, req.FlowJobName)
	srcConn, srcClose, err := connectors.GetByNameAs[connectors.CDCPullConnectorCore](ctx, nil, a.CatalogPool, req.PeerName)
	if err != nil {
		if _, ok := errors.AsType[*exceptions.NotFoundError](err); ok {
			logger := internal.LoggerFromCtx(ctx)
			logger.Warn("peer missing, skipping", slog.String("peer", req.PeerName))
			return nil
		}
		if _, ok := errors.AsType[*exceptions.AuthError](err); ok {
			logger := internal.LoggerFromCtx(ctx)
			logger.Warn("auth error, skipping to avoid triggering security tools", slog.String("peer", req.PeerName))
			return nil
		}
		return a.Alerter.LogFlowError(ctx, req.FlowJobName,
			exceptions.NewDropFlowError(fmt.Errorf("[DropFlowSource] failed to get source connector: %w", err)),
		)
	}
	defer srcClose(ctx)

	if err := srcConn.PullFlowCleanup(ctx, req.FlowJobName); err != nil {
		if dnsErr, ok := errors.AsType[*net.DNSError](err); ok && dnsErr.IsNotFound {
			a.Alerter.LogFlowWarning(ctx, req.FlowJobName, fmt.Errorf("[DropFlowSource] hostname not found, skipping: %w", err))
			return nil
		} else {
			pullCleanupErr := exceptions.NewDropFlowError(fmt.Errorf("[DropFlowSource] failed to clean up source: %w", err))
			if !shared.IsSQLStateError(err, pgerrcode.ObjectInUse) {
				// don't alert when PID active
				_ = a.Alerter.LogFlowError(ctx, req.FlowJobName, pullCleanupErr)
			}
			return pullCleanupErr
		}
	}

	a.Alerter.LogFlowInfo(ctx, req.FlowJobName, "Cleaned up source peer replication objects.")

	return nil
}

func (a *FlowableActivity) DropFlowDestination(ctx context.Context, req *protos.DropFlowActivityInput) error {
	ctx = context.WithValue(ctx, shared.FlowNameKey, req.FlowJobName)