diff --git a/flytectl/cmd/create/execution.go b/flytectl/cmd/create/execution.go
index b687790e12..d80f310b90 100644
--- a/flytectl/cmd/create/execution.go
+++ b/flytectl/cmd/create/execution.go
@@ -72,6 +72,12 @@ The root project and domain flags of -p and -d should point to task/launch plans
 
  flytectl create execution --execFile execution_spec.yaml -p flytectldemo -d development --targetProject flytesnacks
 
+Also an execution can be relaunched by passing in current execution id.
+
+::
+
+ flytectl create execution --relaunch ffb31066a0f8b4d52b77 -p flytectldemo -d development
+
 Usage
 `
 )
@@ -86,6 +92,7 @@ type ExecutionConfig struct {
 	TargetProject   string `json:"targetProject" pflag:",project where execution needs to be created.If not specified configured project would be used."`
 	KubeServiceAcct string `json:"kubeServiceAcct" pflag:",kubernetes service account AuthRole for launching execution."`
 	IamRoleARN      string `json:"iamRoleARN" pflag:",iam role ARN AuthRole for launching execution."`
+	Relaunch        string `json:"relaunch" pflag:",execution id to be relaunched."`
 	// Non plfag section is read from the execution config generated by get task/launchplan
 	Workflow string                 `json:"workflow,omitempty"`
 	Task     string                 `json:"task,omitempty"`
@@ -93,9 +100,17 @@ type ExecutionConfig struct {
 	Inputs   map[string]interface{} `json:"inputs"`
 }
 
+type ExecutionType int
+
+const (
+	Task ExecutionType = iota
+	Workflow
+	Relaunch
+)
+
 type ExecutionParams struct {
-	name   string
-	isTask bool
+	name     string
+	execType ExecutionType
 }
 
 var (
@@ -111,14 +126,19 @@ func createExecutionCommand(ctx context.Context, args []string, cmdCtx cmdCore.C
 		return err
 	}
 	var executionRequest *admin.ExecutionCreateRequest
-	if execParams.isTask {
+	switch execParams.execType {
+	case Relaunch:
+		return relaunchExecution(ctx, execParams.name, sourceProject, sourceDomain, cmdCtx)
+	case Task:
 		if executionRequest, err = createExecutionRequestForTask(ctx, execParams.name, sourceProject, sourceDomain, cmdCtx); err != nil {
 			return err
 		}
-	} else {
+	case Workflow:
 		if executionRequest, err = createExecutionRequestForWorkflow(ctx, execParams.name, sourceProject, sourceDomain, cmdCtx); err != nil {
 			return err
 		}
+	default:
+		return fmt.Errorf("invalid execution type %v", execParams.execType)
 	}
 	exec, _err := cmdCtx.AdminClient().CreateExecution(ctx, executionRequest)
 	if _err != nil {
diff --git a/flytectl/cmd/create/execution_test.go b/flytectl/cmd/create/execution_test.go
index cb9fa4cc15..9a593760c9 100644
--- a/flytectl/cmd/create/execution_test.go
+++ b/flytectl/cmd/create/execution_test.go
@@ -1,6 +1,7 @@
 package create
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/flyteorg/flytectl/cmd/config"
@@ -129,6 +130,36 @@ func createExecutionSetup() {
 	}
 	mockClient.OnGetLaunchPlanMatch(ctx, objectGetRequest).Return(launchPlan1, nil)
 }
+
+func TestCreateTaskExecutionFunc(t *testing.T) {
+	setup()
+	createExecutionSetup()
+	executionCreateResponseTask := &admin.ExecutionCreateResponse{
+		Id: &core.WorkflowExecutionIdentifier{
+			Project: "flytesnacks",
+			Domain:  "development",
+			Name:    "ff513c0e44b5b4a35aa5",
+		},
+	}
+	mockClient.OnCreateExecutionMatch(ctx, mock.Anything).Return(executionCreateResponseTask, nil)
+	executionConfig.ExecFile = testDataFolder + "task_execution_spec.yaml"
+	err = createExecutionCommand(ctx, args, cmdCtx)
+	assert.Nil(t, err)
+	mockClient.AssertCalled(t, "CreateExecution", ctx, mock.Anything)
+	tearDownAndVerify(t, `execution identifier project:"flytesnacks" domain:"development" name:"ff513c0e44b5b4a35aa5" `)
+}
+
+func TestCreateTaskExecutionFuncError(t *testing.T) {
+	setup()
+	createExecutionSetup()
+	mockClient.OnCreateExecutionMatch(ctx, mock.Anything).Return(nil, fmt.Errorf("error launching task"))
+	executionConfig.ExecFile = testDataFolder + "task_execution_spec.yaml"
+	err = createExecutionCommand(ctx, args, cmdCtx)
+	assert.NotNil(t, err)
+	assert.Equal(t, fmt.Errorf("error launching task"), err)
+	mockClient.AssertCalled(t, "CreateExecution", ctx, mock.Anything)
+}
+
 func TestCreateLaunchPlanExecutionFunc(t *testing.T) {
 	setup()
 	createExecutionSetup()
@@ -147,20 +178,46 @@ func TestCreateLaunchPlanExecutionFunc(t *testing.T) {
 	tearDownAndVerify(t, `execution identifier project:"flytesnacks" domain:"development" name:"f652ea3596e7f4d80a0e"`)
 }
 
-func TestCreateTaskExecutionFunc(t *testing.T) {
+func TestCreateRelaunchExecutionFunc(t *testing.T) {
 	setup()
 	createExecutionSetup()
-	executionCreateResponseTask := &admin.ExecutionCreateResponse{
+	relaunchExecResponse := &admin.ExecutionCreateResponse{
 		Id: &core.WorkflowExecutionIdentifier{
 			Project: "flytesnacks",
 			Domain:  "development",
-			Name:    "ff513c0e44b5b4a35aa5",
+			Name:    "f652ea3596e7f4d80a0e",
 		},
 	}
-	mockClient.OnCreateExecutionMatch(ctx, mock.Anything).Return(executionCreateResponseTask, nil)
-	executionConfig.ExecFile = testDataFolder + "task_execution_spec.yaml"
+
+	executionConfig.Relaunch = relaunchExecResponse.Id.Name
+	relaunchRequest := &admin.ExecutionRelaunchRequest{
+		Id: &core.WorkflowExecutionIdentifier{
+			Name:    executionConfig.Relaunch,
+			Project: config.GetConfig().Project,
+			Domain:  config.GetConfig().Domain,
+		},
+	}
+	mockClient.OnRelaunchExecutionMatch(ctx, relaunchRequest).Return(relaunchExecResponse, nil)
 	err = createExecutionCommand(ctx, args, cmdCtx)
 	assert.Nil(t, err)
-	mockClient.AssertCalled(t, "CreateExecution", ctx, mock.Anything)
-	tearDownAndVerify(t, `execution identifier project:"flytesnacks" domain:"development" name:"ff513c0e44b5b4a35aa5" `)
+	mockClient.AssertCalled(t, "RelaunchExecution", ctx, relaunchRequest)
+	tearDownAndVerify(t, `execution identifier project:"flytesnacks" domain:"development" name:"f652ea3596e7f4d80a0e"`)
+}
+
+func TestCreateExecutionFuncInvalid(t *testing.T) {
+	setup()
+	createExecutionSetup()
+	executionConfig.Relaunch = ""
+	executionConfig.ExecFile = ""
+	err = createExecutionCommand(ctx, args, cmdCtx)
+	assert.NotNil(t, err)
+	assert.Equal(t, fmt.Errorf("executionConfig or relaunch can't be empty. Run the flytectl get task/launchplan to generate the config"), err)
+	executionConfig.ExecFile = "Invalid-file"
+	err = createExecutionCommand(ctx, args, cmdCtx)
+	assert.NotNil(t, err)
+	assert.Equal(t, fmt.Errorf("unable to read from %v yaml file", executionConfig.ExecFile), err)
+	executionConfig.ExecFile = testDataFolder + "invalid_execution_spec.yaml"
+	err = createExecutionCommand(ctx, args, cmdCtx)
+	assert.NotNil(t, err)
+	assert.Equal(t, fmt.Errorf("either one of task or workflow name should be specified to launch an execution"), err)
 }
diff --git a/flytectl/cmd/create/execution_util.go b/flytectl/cmd/create/execution_util.go
index acb3f595d8..dd21867ba8 100644
--- a/flytectl/cmd/create/execution_util.go
+++ b/flytectl/cmd/create/execution_util.go
@@ -2,7 +2,6 @@ package create
 
 import (
 	"context"
-	"errors"
 	"fmt"
 	"io/ioutil"
 	"strings"
@@ -11,7 +10,6 @@ import (
 	cmdGet "github.com/flyteorg/flytectl/cmd/get"
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
-
 	"github.com/google/uuid"
 	"sigs.k8s.io/yaml"
 )
@@ -20,7 +18,7 @@ func createExecutionRequestForWorkflow(ctx context.Context, workflowName string,
 	var lp *admin.LaunchPlan
 	var err error
 	// Fetch the launch plan
-	if lp, err = cmdGet.FetchLPVersion(ctx, workflowName, executionConfig.Version, project, domain, cmdCtx); err != nil {
+	if lp, err = cmdGet.DefaultFetcher.FetchLPVersion(ctx, workflowName, executionConfig.Version, project, domain, cmdCtx); err != nil {
 		return nil, err
 	}
 	// Create workflow params literal map
@@ -70,6 +68,21 @@ func createExecutionRequestForTask(ctx context.Context, taskName string, project
 	return createExecutionRequest(ID, inputs, authRole), nil
 }
 
+func relaunchExecution(ctx context.Context, executionName string, project string, domain string, cmdCtx cmdCore.CommandContext) error {
+	relaunchedExec, err := cmdCtx.AdminClient().RelaunchExecution(ctx, &admin.ExecutionRelaunchRequest{
+		Id: &core.WorkflowExecutionIdentifier{
+			Name:    executionName,
+			Project: project,
+			Domain:  domain,
+		},
+	})
+	if err != nil {
+		return err
+	}
+	fmt.Printf("execution identifier %v\n", relaunchedExec.Id)
+	return nil
+}
+
 func createExecutionRequest(ID *core.Identifier, inputs *core.LiteralMap, authRole *admin.AuthRole) *admin.ExecutionCreateRequest {
 	return &admin.ExecutionCreateRequest{
 		Project: executionConfig.TargetProject,
@@ -100,32 +113,36 @@ func readExecConfigFromFile(fileName string) (*ExecutionConfig, error) {
 	return &executionConfigRead, nil
 }
 
-func resolveOverrides(readExecutionConfig *ExecutionConfig, project string, domain string) {
+func resolveOverrides(toBeOverridden *ExecutionConfig, project string, domain string) {
 	if executionConfig.KubeServiceAcct != "" {
-		readExecutionConfig.KubeServiceAcct = executionConfig.KubeServiceAcct
+		toBeOverridden.KubeServiceAcct = executionConfig.KubeServiceAcct
 	}
 	if executionConfig.IamRoleARN != "" {
-		readExecutionConfig.IamRoleARN = executionConfig.IamRoleARN
+		toBeOverridden.IamRoleARN = executionConfig.IamRoleARN
 	}
 	if executionConfig.TargetProject != "" {
-		readExecutionConfig.TargetProject = executionConfig.TargetProject
+		toBeOverridden.TargetProject = executionConfig.TargetProject
 	}
 	if executionConfig.TargetDomain != "" {
-		readExecutionConfig.TargetDomain = executionConfig.TargetDomain
+		toBeOverridden.TargetDomain = executionConfig.TargetDomain
 	}
 	// Use the root project and domain to launch the task/workflow if target is unspecified
 	if executionConfig.TargetProject == "" {
-		readExecutionConfig.TargetProject = project
+		toBeOverridden.TargetProject = project
 	}
 	if executionConfig.TargetDomain == "" {
-		readExecutionConfig.TargetDomain = domain
+		toBeOverridden.TargetDomain = domain
 	}
 }
 
 func readConfigAndValidate(project string, domain string) (ExecutionParams, error) {
 	executionParams := ExecutionParams{}
-	if executionConfig.ExecFile == "" {
-		return executionParams, errors.New("executionConfig can't be empty. Run the flytectl get task/launchplan to generate the config")
+	if executionConfig.ExecFile == "" && executionConfig.Relaunch == "" {
+		return executionParams, fmt.Errorf("executionConfig or relaunch can't be empty. Run the flytectl get task/launchplan to generate the config")
+	}
+	if executionConfig.Relaunch != "" {
+		resolveOverrides(executionConfig, project, domain)
+		return ExecutionParams{name: executionConfig.Relaunch, execType: Relaunch}, nil
 	}
 	var readExecutionConfig *ExecutionConfig
 	var err error
@@ -138,11 +155,13 @@ func readConfigAndValidate(project string, domain string) (ExecutionParams, erro
 	isTask := readExecutionConfig.Task != ""
 	isWorkflow := readExecutionConfig.Workflow != ""
 	if isTask == isWorkflow {
-		return executionParams, errors.New("either one of task or workflow name should be specified to launch an execution")
+		return executionParams, fmt.Errorf("either one of task or workflow name should be specified to launch an execution")
 	}
 	name := readExecutionConfig.Task
+	execType := Task
 	if !isTask {
 		name = readExecutionConfig.Workflow
+		execType = Workflow
 	}
-	return ExecutionParams{name: name, isTask: isTask}, nil
+	return ExecutionParams{name: name, execType: execType}, nil
 }
diff --git a/flytectl/cmd/create/execution_util_test.go b/flytectl/cmd/create/execution_util_test.go
new file mode 100644
index 0000000000..db3a084225
--- /dev/null
+++ b/flytectl/cmd/create/execution_util_test.go
@@ -0,0 +1,56 @@
+package create
+
+import (
+	"errors"
+	"testing"
+
+	"github.com/flyteorg/flytectl/cmd/config"
+	"github.com/flyteorg/flytectl/cmd/testutils"
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
+
+	"github.com/stretchr/testify/assert"
+)
+
+var (
+	relaunchExecResponse *admin.ExecutionCreateResponse
+	relaunchRequest      *admin.ExecutionRelaunchRequest
+)
+
+// This function needs to be called after testutils.Steup()
+func createExecutionUtilSetup() {
+	ctx = testutils.Ctx
+	cmdCtx = testutils.CmdCtx
+	mockClient = testutils.MockClient
+	relaunchExecResponse = &admin.ExecutionCreateResponse{
+		Id: &core.WorkflowExecutionIdentifier{
+			Project: "flytesnacks",
+			Domain:  "development",
+			Name:    "f652ea3596e7f4d80a0e",
+		},
+	}
+	relaunchRequest = &admin.ExecutionRelaunchRequest{
+		Id: &core.WorkflowExecutionIdentifier{
+			Name:    "execName",
+			Project: config.GetConfig().Project,
+			Domain:  config.GetConfig().Domain,
+		},
+	}
+}
+
+func TestCreateExecutionForRelaunch(t *testing.T) {
+	setup()
+	createExecutionUtilSetup()
+	mockClient.OnRelaunchExecutionMatch(ctx, relaunchRequest).Return(relaunchExecResponse, nil)
+	err = relaunchExecution(ctx, "execName", config.GetConfig().Project, config.GetConfig().Domain, cmdCtx)
+	assert.Nil(t, err)
+}
+
+func TestCreateExecutionForRelaunchNotFound(t *testing.T) {
+	setup()
+	createExecutionUtilSetup()
+	mockClient.OnRelaunchExecutionMatch(ctx, relaunchRequest).Return(nil, errors.New("unknown execution"))
+	err = relaunchExecution(ctx, "execName", config.GetConfig().Project, config.GetConfig().Domain, cmdCtx)
+	assert.NotNil(t, err)
+	assert.Equal(t, err, errors.New("unknown execution"))
+}
diff --git a/flytectl/cmd/create/executionconfig_flags.go b/flytectl/cmd/create/executionconfig_flags.go
index 4a9774dff0..59dc9adf41 100755
--- a/flytectl/cmd/create/executionconfig_flags.go
+++ b/flytectl/cmd/create/executionconfig_flags.go
@@ -41,10 +41,12 @@ func (ExecutionConfig) mustMarshalJSON(v json.Marshaler) string {
 // flags is json-name.json-sub-name... etc.
 func (cfg ExecutionConfig) GetPFlagSet(prefix string) *pflag.FlagSet {
 	cmdFlags := pflag.NewFlagSet("ExecutionConfig", pflag.ExitOnError)
-	cmdFlags.StringVar(&(executionConfig.ExecFile),fmt.Sprintf("%v%v", prefix, "execFile"), executionConfig.ExecFile, "file for the execution params.If not specified defaults to <<workflow/task>_name>.execution_spec.yaml")
-	cmdFlags.StringVar(&(executionConfig.TargetDomain),fmt.Sprintf("%v%v", prefix, "targetDomain"), executionConfig.TargetDomain, "project where execution needs to be created.If not specified configured domain would be used.")
-	cmdFlags.StringVar(&(executionConfig.TargetProject),fmt.Sprintf("%v%v", prefix, "targetProject"), executionConfig.TargetProject, "project where execution needs to be created.If not specified configured project would be used.")
-	cmdFlags.StringVar(&(executionConfig.KubeServiceAcct),fmt.Sprintf("%v%v", prefix, "kubeServiceAcct"), executionConfig.KubeServiceAcct, "kubernetes service account AuthRole for launching execution.")
-	cmdFlags.StringVar(&(executionConfig.IamRoleARN),fmt.Sprintf("%v%v", prefix, "iamRoleARN"), executionConfig.IamRoleARN, "iam role ARN AuthRole for launching execution.")
+	cmdFlags.StringVar(&(executionConfig.ExecFile), fmt.Sprintf("%v%v", prefix, "execFile"), executionConfig.ExecFile, "file for the execution params.If not specified defaults to <<workflow/task>_name>.execution_spec.yaml")
+	cmdFlags.StringVar(&(executionConfig.TargetDomain), fmt.Sprintf("%v%v", prefix, "targetDomain"), executionConfig.TargetDomain, "project where execution needs to be created.If not specified configured domain would be used.")
+	cmdFlags.StringVar(&(executionConfig.TargetProject), fmt.Sprintf("%v%v", prefix, "targetProject"), executionConfig.TargetProject, "project where execution needs to be created.If not specified configured project would be used.")
+	cmdFlags.StringVar(&(executionConfig.KubeServiceAcct), fmt.Sprintf("%v%v", prefix, "kubeServiceAcct"), executionConfig.KubeServiceAcct, "kubernetes service account AuthRole for launching execution.")
+	cmdFlags.StringVar(&(executionConfig.IamRoleARN), fmt.Sprintf("%v%v", prefix, "iamRoleARN"), executionConfig.IamRoleARN, "iam role ARN AuthRole for launching execution.")
+	cmdFlags.StringVar(&(executionConfig.Relaunch), fmt.Sprintf("%v%v", prefix, "relaunch"), executionConfig.Relaunch, "execution id to be relaunched.")
+
 	return cmdFlags
 }
diff --git a/flytectl/cmd/get/execution.go b/flytectl/cmd/get/execution.go
index f7d91812c1..e6ab431df5 100644
--- a/flytectl/cmd/get/execution.go
+++ b/flytectl/cmd/get/execution.go
@@ -4,7 +4,6 @@ import (
 	"context"
 
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
-	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/core"
 	"github.com/flyteorg/flytestdlib/logger"
 	"github.com/golang/protobuf/proto"
 
@@ -70,13 +69,7 @@ func getExecutionFunc(ctx context.Context, args []string, cmdCtx cmdCore.Command
 	var executions []*admin.Execution
 	if len(args) > 0 {
 		name := args[0]
-		execution, err := cmdCtx.AdminClient().GetExecution(ctx, &admin.WorkflowExecutionGetRequest{
-			Id: &core.WorkflowExecutionIdentifier{
-				Project: config.GetConfig().Project,
-				Domain:  config.GetConfig().Domain,
-				Name:    name,
-			},
-		})
+		execution, err := DefaultFetcher.FetchExecution(ctx, name, config.GetConfig().Project, config.GetConfig().Domain, cmdCtx)
 		if err != nil {
 			return err
 		}
diff --git a/flytectl/cmd/get/execution_util.go b/flytectl/cmd/get/execution_util.go
index da330af39e..2a563f4165 100644
--- a/flytectl/cmd/get/execution_util.go
+++ b/flytectl/cmd/get/execution_util.go
@@ -1,11 +1,13 @@
 package get
 
 import (
+	"context"
 	"errors"
 	"fmt"
 	"io/ioutil"
 	"os"
 
+	cmdCore "github.com/flyteorg/flytectl/cmd/core"
 	cmdUtil "github.com/flyteorg/flytectl/pkg/commandutils"
 	"github.com/flyteorg/flyteidl/clients/go/coreutils"
 	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
@@ -27,6 +29,20 @@ type ExecutionConfig struct {
 	Inputs          map[string]interface{} `json:"inputs"`
 }
 
+func (f FetcherImpl) FetchExecution(ctx context.Context, name string, project string, domain string, cmdCtx cmdCore.CommandContext) (*admin.Execution, error) {
+	e, err := cmdCtx.AdminClient().GetExecution(ctx, &admin.WorkflowExecutionGetRequest{
+		Id: &core.WorkflowExecutionIdentifier{
+			Project: project,
+			Domain:  domain,
+			Name:    name,
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+	return e, nil
+}
+
 func WriteExecConfigToFile(executionConfig ExecutionConfig, fileName string) error {
 	d, err := yaml.Marshal(executionConfig)
 	if err != nil {
diff --git a/flytectl/cmd/get/get.go b/flytectl/cmd/get/get.go
index e875a0ca01..46f392c4d0 100644
--- a/flytectl/cmd/get/get.go
+++ b/flytectl/cmd/get/get.go
@@ -2,6 +2,7 @@ package get
 
 import (
 	cmdcore "github.com/flyteorg/flytectl/cmd/core"
+	"github.com/flyteorg/flytectl/cmd/get/interfaces"
 
 	"github.com/spf13/cobra"
 )
@@ -17,6 +18,17 @@ Example get projects.
 `
 )
 
+var (
+	DefaultFetcher = NewFetcherImpl()
+)
+
+func NewFetcherImpl() interfaces.Fetcher {
+	return FetcherImpl{}
+}
+
+type FetcherImpl struct {
+}
+
 // CreateGetCommand will return get command
 func CreateGetCommand() *cobra.Command {
 	getCmd := &cobra.Command{
diff --git a/flytectl/cmd/get/interfaces/fetcher_interface.go b/flytectl/cmd/get/interfaces/fetcher_interface.go
new file mode 100644
index 0000000000..62a007e848
--- /dev/null
+++ b/flytectl/cmd/get/interfaces/fetcher_interface.go
@@ -0,0 +1,16 @@
+package interfaces
+
+import (
+	"context"
+
+	cmdCore "github.com/flyteorg/flytectl/cmd/core"
+	"github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+)
+
+//go:generate mockery -all -case=underscore
+
+// Interface for exposing the fetch capabilities to other modules. eg : create execution which requires to fetch launchplan details.
+type Fetcher interface {
+	FetchExecution(ctx context.Context, name string, project string, domain string, cmdCtx cmdCore.CommandContext) (*admin.Execution, error)
+	FetchLPVersion(ctx context.Context, name string, version string, project string, domain string, cmdCtx cmdCore.CommandContext) (*admin.LaunchPlan, error)
+}
diff --git a/flytectl/cmd/get/interfaces/mocks/fetcher.go b/flytectl/cmd/get/interfaces/mocks/fetcher.go
new file mode 100644
index 0000000000..61742c70c6
--- /dev/null
+++ b/flytectl/cmd/get/interfaces/mocks/fetcher.go
@@ -0,0 +1,99 @@
+// Code generated by mockery v1.0.1. DO NOT EDIT.
+
+package mocks
+
+import (
+	cmdcore "github.com/flyteorg/flytectl/cmd/core"
+	admin "github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+)
+
+// Fetcher is an autogenerated mock type for the Fetcher type
+type Fetcher struct {
+	mock.Mock
+}
+
+type Fetcher_FetchExecution struct {
+	*mock.Call
+}
+
+func (_m Fetcher_FetchExecution) Return(_a0 *admin.Execution, _a1 error) *Fetcher_FetchExecution {
+	return &Fetcher_FetchExecution{Call: _m.Call.Return(_a0, _a1)}
+}
+
+func (_m *Fetcher) OnFetchExecution(ctx context.Context, name string, project string, domain string, cmdCtx cmdcore.CommandContext) *Fetcher_FetchExecution {
+	c := _m.On("FetchExecution", ctx, name, project, domain, cmdCtx)
+	return &Fetcher_FetchExecution{Call: c}
+}
+
+func (_m *Fetcher) OnFetchExecutionMatch(matchers ...interface{}) *Fetcher_FetchExecution {
+	c := _m.On("FetchExecution", matchers...)
+	return &Fetcher_FetchExecution{Call: c}
+}
+
+// FetchExecution provides a mock function with given fields: ctx, name, project, domain, cmdCtx
+func (_m *Fetcher) FetchExecution(ctx context.Context, name string, project string, domain string, cmdCtx cmdcore.CommandContext) (*admin.Execution, error) {
+	ret := _m.Called(ctx, name, project, domain, cmdCtx)
+
+	var r0 *admin.Execution
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, string, cmdcore.CommandContext) *admin.Execution); ok {
+		r0 = rf(ctx, name, project, domain, cmdCtx)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*admin.Execution)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, string, string, cmdcore.CommandContext) error); ok {
+		r1 = rf(ctx, name, project, domain, cmdCtx)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
+
+type Fetcher_FetchLPVersion struct {
+	*mock.Call
+}
+
+func (_m Fetcher_FetchLPVersion) Return(_a0 *admin.LaunchPlan, _a1 error) *Fetcher_FetchLPVersion {
+	return &Fetcher_FetchLPVersion{Call: _m.Call.Return(_a0, _a1)}
+}
+
+func (_m *Fetcher) OnFetchLPVersion(ctx context.Context, name string, version string, project string, domain string, cmdCtx cmdcore.CommandContext) *Fetcher_FetchLPVersion {
+	c := _m.On("FetchLPVersion", ctx, name, version, project, domain, cmdCtx)
+	return &Fetcher_FetchLPVersion{Call: c}
+}
+
+func (_m *Fetcher) OnFetchLPVersionMatch(matchers ...interface{}) *Fetcher_FetchLPVersion {
+	c := _m.On("FetchLPVersion", matchers...)
+	return &Fetcher_FetchLPVersion{Call: c}
+}
+
+// FetchLPVersion provides a mock function with given fields: ctx, name, version, project, domain, cmdCtx
+func (_m *Fetcher) FetchLPVersion(ctx context.Context, name string, version string, project string, domain string, cmdCtx cmdcore.CommandContext) (*admin.LaunchPlan, error) {
+	ret := _m.Called(ctx, name, version, project, domain, cmdCtx)
+
+	var r0 *admin.LaunchPlan
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, string, string, cmdcore.CommandContext) *admin.LaunchPlan); ok {
+		r0 = rf(ctx, name, version, project, domain, cmdCtx)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*admin.LaunchPlan)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, string, string, string, cmdcore.CommandContext) error); ok {
+		r1 = rf(ctx, name, version, project, domain, cmdCtx)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
diff --git a/flytectl/cmd/get/interfaces/mocks/get_execution.go b/flytectl/cmd/get/interfaces/mocks/get_execution.go
new file mode 100644
index 0000000000..0df4fcf5ac
--- /dev/null
+++ b/flytectl/cmd/get/interfaces/mocks/get_execution.go
@@ -0,0 +1,58 @@
+// Code generated by mockery v1.0.1. DO NOT EDIT.
+
+package mocks
+
+import (
+	cmdcore "github.com/flyteorg/flytectl/cmd/core"
+	admin "github.com/flyteorg/flyteidl/gen/pb-go/flyteidl/admin"
+
+	context "context"
+
+	mock "github.com/stretchr/testify/mock"
+)
+
+// GetExecution is an autogenerated mock type for the GetExecution type
+type GetExecution struct {
+	mock.Mock
+}
+
+type GetExecution_FetchExecution struct {
+	*mock.Call
+}
+
+func (_m GetExecution_FetchExecution) Return(_a0 *admin.Execution, _a1 error) *GetExecution_FetchExecution {
+	return &GetExecution_FetchExecution{Call: _m.Call.Return(_a0, _a1)}
+}
+
+func (_m *GetExecution) OnFetchExecution(ctx context.Context, name string, project string, domain string, cmdCtx cmdcore.CommandContext) *GetExecution_FetchExecution {
+	c := _m.On("FetchExecution", ctx, name, project, domain, cmdCtx)
+	return &GetExecution_FetchExecution{Call: c}
+}
+
+func (_m *GetExecution) OnFetchExecutionMatch(matchers ...interface{}) *GetExecution_FetchExecution {
+	c := _m.On("FetchExecution", matchers...)
+	return &GetExecution_FetchExecution{Call: c}
+}
+
+// FetchExecution provides a mock function with given fields: ctx, name, project, domain, cmdCtx
+func (_m *GetExecution) FetchExecution(ctx context.Context, name string, project string, domain string, cmdCtx cmdcore.CommandContext) (*admin.Execution, error) {
+	ret := _m.Called(ctx, name, project, domain, cmdCtx)
+
+	var r0 *admin.Execution
+	if rf, ok := ret.Get(0).(func(context.Context, string, string, string, cmdcore.CommandContext) *admin.Execution); ok {
+		r0 = rf(ctx, name, project, domain, cmdCtx)
+	} else {
+		if ret.Get(0) != nil {
+			r0 = ret.Get(0).(*admin.Execution)
+		}
+	}
+
+	var r1 error
+	if rf, ok := ret.Get(1).(func(context.Context, string, string, string, cmdcore.CommandContext) error); ok {
+		r1 = rf(ctx, name, project, domain, cmdCtx)
+	} else {
+		r1 = ret.Error(1)
+	}
+
+	return r0, r1
+}
diff --git a/flytectl/cmd/get/launch_plan_util.go b/flytectl/cmd/get/launch_plan_util.go
index e72c6e9094..d3a803e32b 100644
--- a/flytectl/cmd/get/launch_plan_util.go
+++ b/flytectl/cmd/get/launch_plan_util.go
@@ -20,7 +20,7 @@ func FetchLPForName(ctx context.Context, name string, project string, domain str
 		}
 		launchPlans = append(launchPlans, lp)
 	} else if launchPlanConfig.Version != "" {
-		if lp, err = FetchLPVersion(ctx, name, launchPlanConfig.Version, project, domain, cmdCtx); err != nil {
+		if lp, err = DefaultFetcher.FetchLPVersion(ctx, name, launchPlanConfig.Version, project, domain, cmdCtx); err != nil {
 			return nil, err
 		}
 		launchPlans = append(launchPlans, lp)
@@ -73,7 +73,7 @@ func FetchLPLatestVersion(ctx context.Context, name string, project string, doma
 	return lp, nil
 }
 
-func FetchLPVersion(ctx context.Context, name string, version string, project string, domain string, cmdCtx cmdCore.CommandContext) (*admin.LaunchPlan, error) {
+func (f FetcherImpl) FetchLPVersion(ctx context.Context, name string, version string, project string, domain string, cmdCtx cmdCore.CommandContext) (*admin.LaunchPlan, error) {
 	lp, err := cmdCtx.AdminClient().GetLaunchPlan(ctx, &admin.ObjectGetRequest{
 		Id: &core.Identifier{
 			ResourceType: core.ResourceType_LAUNCH_PLAN,
diff --git a/flytectl/cmd/get/launchplanconfig_flags.go b/flytectl/cmd/get/launchplanconfig_flags.go
index 00becdd073..f4e37ea0c4 100755
--- a/flytectl/cmd/get/launchplanconfig_flags.go
+++ b/flytectl/cmd/get/launchplanconfig_flags.go
@@ -43,6 +43,6 @@ func (cfg LaunchPlanConfig) GetPFlagSet(prefix string) *pflag.FlagSet {
 	cmdFlags := pflag.NewFlagSet("LaunchPlanConfig", pflag.ExitOnError)
 	cmdFlags.StringVar(&(launchPlanConfig.ExecFile), fmt.Sprintf("%v%v", prefix, "execFile"), launchPlanConfig.ExecFile, "execution file name to be used for generating execution spec of a single launchplan.")
 	cmdFlags.StringVar(&(launchPlanConfig.Version), fmt.Sprintf("%v%v", prefix, "version"), launchPlanConfig.Version, "version of the launchplan to be fetched.")
-	cmdFlags.BoolVar(&(launchPlanConfig.Latest),fmt.Sprintf("%v%v", prefix, "latest"), launchPlanConfig.Latest, "flag to indicate to fetch the latest version, version flag will be ignored in this case")
+	cmdFlags.BoolVar(&(launchPlanConfig.Latest), fmt.Sprintf("%v%v", prefix, "latest"), launchPlanConfig.Latest, "flag to indicate to fetch the latest version, version flag will be ignored in this case")
 	return cmdFlags
 }
diff --git a/flytectl/cmd/get/taskconfig_flags.go b/flytectl/cmd/get/taskconfig_flags.go
index 5379d61df7..71a3fadabc 100755
--- a/flytectl/cmd/get/taskconfig_flags.go
+++ b/flytectl/cmd/get/taskconfig_flags.go
@@ -41,8 +41,8 @@ func (TaskConfig) mustMarshalJSON(v json.Marshaler) string {
 // flags is json-name.json-sub-name... etc.
 func (cfg TaskConfig) GetPFlagSet(prefix string) *pflag.FlagSet {
 	cmdFlags := pflag.NewFlagSet("TaskConfig", pflag.ExitOnError)
-	cmdFlags.StringVar(&(taskConfig.ExecFile),fmt.Sprintf("%v%v", prefix, "execFile"), taskConfig.ExecFile, "execution file name to be used for generating execution spec of a single task.")
-	cmdFlags.StringVar(&(taskConfig.Version),fmt.Sprintf("%v%v", prefix, "version"), taskConfig.Version, "version of the task to be fetched.")
-	cmdFlags.BoolVar(&(taskConfig.Latest),fmt.Sprintf("%v%v", prefix, "latest"), taskConfig.Latest, "flag to indicate to fetch the latest version, version flag will be ignored in this case")
+	cmdFlags.StringVar(&(taskConfig.ExecFile), fmt.Sprintf("%v%v", prefix, "execFile"), taskConfig.ExecFile, "execution file name to be used for generating execution spec of a single task.")
+	cmdFlags.StringVar(&(taskConfig.Version), fmt.Sprintf("%v%v", prefix, "version"), taskConfig.Version, "version of the task to be fetched.")
+	cmdFlags.BoolVar(&(taskConfig.Latest), fmt.Sprintf("%v%v", prefix, "latest"), taskConfig.Latest, "flag to indicate to fetch the latest version, version flag will be ignored in this case")
 	return cmdFlags
 }
diff --git a/flytectl/cmd/testdata/invalid_execution_spec.yaml b/flytectl/cmd/testdata/invalid_execution_spec.yaml
new file mode 100644
index 0000000000..556bb512c7
--- /dev/null
+++ b/flytectl/cmd/testdata/invalid_execution_spec.yaml
@@ -0,0 +1,12 @@
+iamRoleARN: ""
+inputs:
+  numbers:
+  - 0
+  numbers_count: 0
+  run_local_at_count: 10
+kubeServiceAcct: ""
+targetDomain: ""
+targetProject: ""
+version: v3
+workflow: core.advanced.run_merge_sort.merge_sort
+task: core.advanced.run_merge_sort.merge
diff --git a/flytectl/cmd/testdata/task_execution_spec.yaml b/flytectl/cmd/testdata/task_execution_spec.yaml
index aba4e46d6d..aa33cdb355 100644
--- a/flytectl/cmd/testdata/task_execution_spec.yaml
+++ b/flytectl/cmd/testdata/task_execution_spec.yaml
@@ -1,4 +1,4 @@
-iamRoleARN: ""
+iamRoleARN: "iamRoleARN"
 inputs:
   sorted_list1:
   - 0
@@ -8,7 +8,7 @@ inputs:
   - 1
   - 3
   - 5
-kubeServiceAcct: ""
+kubeServiceAcct: "kubeServiceAcct"
 targetDomain: "development"
 targetProject: "flytesnacks"
 task: core.advanced.run_merge_sort.merge
diff --git a/flytectl/docs/source/gen/flytectl_create_execution.rst b/flytectl/docs/source/gen/flytectl_create_execution.rst
index 01b7f0e2b1..5487019357 100644
--- a/flytectl/docs/source/gen/flytectl_create_execution.rst
+++ b/flytectl/docs/source/gen/flytectl_create_execution.rst
@@ -70,6 +70,12 @@ The root project and domain flags of -p and -d should point to task/launch plans
 
  flytectl create execution --execFile execution_spec.yaml -p flytectldemo -d development --targetProject flytesnacks
 
+Also an execution can be relaunched by passing in current execution id.
+
+::
+
+ flytectl create execution --relaunch ffb31066a0f8b4d52b77 -p flytectldemo -d development
+
 Usage
 
 
@@ -86,6 +92,7 @@ Options
   -h, --help                     help for execution
       --iamRoleARN string        iam role ARN AuthRole for launching execution.
       --kubeServiceAcct string   kubernetes service account AuthRole for launching execution.
+      --relaunch string          execution id to be relaunched.
       --targetDomain string      project where execution needs to be created.If not specified configured domain would be used.
       --targetProject string     project where execution needs to be created.If not specified configured project would be used.