Method dropFlow

flow/cmd/handler.go:301–344

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

}

func (h *FlowRequestHandler) dropFlow(
	ctx context.Context,
	flowJobName string,
	deleteStats bool,
) error {
	logs := slog.Group("shutdown-log", slog.String(string(shared.FlowNameKey), flowJobName))

	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),
	}

	if dropFlowHandle, err := h.temporalClient.ExecuteWorkflow(ctx, workflowOptions, peerflow.DropFlowWorkflow, &protos.DropFlowInput{
		FlowJobName:           flowJobName,
		DropFlowStats:         deleteStats,
		FlowConnectionConfigs: pconv.FlowConnectionConfigsToCore(cdcConfig, 0),
		SkipDestinationDrop:   true,
		SkipSourceDrop:        true,
	}); 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)
	} else if err := dropFlowHandle.Get(ctx, nil); err != nil {
		slog.ErrorContext(ctx, "DropFlow workflow did not execute successfully", logs, slog.Any("error", err))
		return fmt.Errorf("DropFlow workflow did not execute successfully: %w", err)
	}

	return nil
}

func (h *FlowRequestHandler) shutdownFlow(
	ctx context.Context,