Method shutdownFlow

flow/cmd/handler.go:346–421

(
	ctx context.Context,
	flowJobName string,
	deleteStats bool,
	skipDestinationDrop bool,
)

}

func (h *FlowRequestHandler) shutdownFlow(
	ctx context.Context,
	flowJobName string,
	deleteStats bool,
	skipDestinationDrop bool,
) error {
	workflowID, err := h.getWorkflowID(ctx, flowJobName)
	if err != nil {
		return err
	}

	logs := slog.Group("shutdown-log",
		slog.String(string(shared.FlowNameKey), flowJobName),
		slog.String("workflowId", workflowID),
	)

	if err := h.handleCancelWorkflow(ctx, workflowID, ""); err != nil {
		slog.ErrorContext(ctx, "unable to cancel workflow", logs, slog.Any("error", err))
		return fmt.Errorf("unable to wait for PeerFlow workflow to close: %w", err)
	}

	isCdc, err := h.isCDCFlow(ctx, flowJobName)
	if err != nil {
		slog.ErrorContext(ctx, "unable to check if workflow is cdc", logs, slog.Any("error", err))
		return fmt.Errorf("unable to determine if workflow is cdc: %w", err)
	}
	var cdcConfig *protos.FlowConnectionConfigs
	if isCdc {
		cdcConfig, err = h.getFlowConfigFromCatalog(ctx, flowJobName)
		if err != nil {
			slog.ErrorContext(ctx, "unable to get cdc config from catalog", logs, slog.Any("error", err))
			return fmt.Errorf("unable to get cdc config from catalog: %w", err)
		}
	}
	dropFlowWorkflowID := fmt.Sprintf("%s-dropflow-%s", flowJobName, uuid.New())
	workflowOptions := client.StartWorkflowOptions{
		ID:                    dropFlowWorkflowID,
		TaskQueue:             h.peerflowTaskQueueID,
		TypedSearchAttributes: shared.NewSearchAttributes(flowJobName),
	}

	dropFlowHandle, err := h.temporalClient.ExecuteWorkflow(ctx, workflowOptions, peerflow.DropFlowWorkflow, &protos.DropFlowInput{
		FlowJobName:           flowJobName,
		DropFlowStats:         deleteStats,
		FlowConnectionConfigs: pconv.FlowConnectionConfigsToCore(cdcConfig, 0),
		SkipDestinationDrop:   skipDestinationDrop,
		// NOTE: Resync is false here during snapshot-only resync
	})
	if err != nil {
		slog.ErrorContext(ctx, "unable to start DropFlow workflow", logs, slog.Any("error", err))
		return fmt.Errorf("unable to start DropFlow workflow: %w", err)
	}

	cancelCtx, cancel := context.WithTimeout(ctx, 5*time.Minute)
	defer cancel()

	errLatch := concurrency.NewLatch[error]()
	go func() {