[KFS-1863] add local mode to flink shell (#2713)

internal/flink/command_shell.go

@@ -11,16 +11,13 @@ import (
 	pcmd "github.com/confluentinc/cli/v3/pkg/cmd"
 	"github.com/confluentinc/cli/v3/pkg/config"
 	"github.com/confluentinc/cli/v3/pkg/errors"
+	"github.com/confluentinc/cli/v3/pkg/featureflags"
 	client "github.com/confluentinc/cli/v3/pkg/flink/app"
-	"github.com/confluentinc/cli/v3/pkg/flink/test/mock"
 	"github.com/confluentinc/cli/v3/pkg/flink/types"
 	"github.com/confluentinc/cli/v3/pkg/log"
 	ppanic "github.com/confluentinc/cli/v3/pkg/panic-recovery"
 )
 
-// If we set this const useFakeGateway to true, we start the client with a simulated gateway client that returns fake data. This is used for debugging.
-const useFakeGateway = false
-
 func (c *command) newShellCommand(prerunner pcmd.PreRunner) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "shell",
@@ -36,6 +33,11 @@ func (c *command) newShellCommand(prerunner pcmd.PreRunner) *cobra.Command {
 	pcmd.AddEnvironmentFlag(cmd, c.AuthenticatedCLICommand)
 	pcmd.AddContextFlag(cmd, c.CLICommand)
 
+	if featureflags.Manager.BoolVariation("cli.flink.internal", c.Context, config.CliLaunchDarklyClient, true, false) {
+		cmd.Flags().StringSlice("config-key", []string{}, "App option keys for local mode.")
+		cmd.Flags().StringSlice("config-value", []string{}, "App option values for local mode.")
+	}
+
 	return cmd
 }
 
@@ -70,14 +72,21 @@ func (c *command) authenticated(authenticated func(*cobra.Command, []string) err
 }
 
 func (c *command) startFlinkSqlClient(prerunner pcmd.PreRunner, cmd *cobra.Command) error {
-	if useFakeGateway {
-		return client.StartApp(
-			mock.NewFakeFlinkGatewayClient(),
-			func() error { return nil },
-			types.ApplicationOptions{
-				Context:   c.Context,
-				UserAgent: c.Version.UserAgent,
-			}, func() {})
+	if featureflags.Manager.BoolVariation("cli.flink.internal", c.Context, config.CliLaunchDarklyClient, true, false) {
+		// get config keys and values from flags
+		configKeys, err := cmd.Flags().GetStringSlice("config-key")
+		if err != nil {
+			return err
+		}
+		configValues, err := cmd.Flags().GetStringSlice("config-value")
+		if err != nil {
+			return err
+		}
+
+		// if configs were passed, we should enter local mode
+		if len(configKeys) > 0 && len(configValues) > 0 {
+			return c.startWithLocalMode(configKeys, configValues)
+		}
 	}
 
 	environmentId, err := cmd.Flags().GetString("environment")
@@ -168,6 +177,24 @@ func (c *command) startFlinkSqlClient(prerunner pcmd.PreRunner, cmd *cobra.Comma
 	return client.StartApp(flinkGatewayClient, c.authenticated(prerunner.Authenticated(c.AuthenticatedCLICommand), cmd, jwtValidator), opts, reportUsage(cmd, c.Config, unsafeTrace))
 }
 
+func (c *command) startWithLocalMode(configKeys, configValues []string) error {
+	// parse app options from given flags
+	appOptions, err := types.ParseApplicationOptionsFromSlices(configKeys, configValues)
+	if err != nil {
+		return err
+	}
+
+	// validate app options
+	if err := appOptions.Validate(); err != nil {
+		return err
+	}
+
+	gatewayClient := ccloudv2.NewFlinkGatewayClient(appOptions.GetGatewayUrl(), c.Version.UserAgent, appOptions.GetUnsafeTrace(), "authToken")
+
+	appOptions.Context = c.Context
+	return client.StartApp(gatewayClient, func() error { return nil }, *appOptions, func() {})
+}
+
 func (c *command) getFlinkLanguageServiceUrl(gatewayClient *ccloudv2.FlinkGatewayClient) (string, error) {
 	if cfg := gatewayClient.GetConfig(); cfg != nil && len(cfg.Servers) > 0 {
 		gatewayUrl := cfg.Servers[0].URL

pkg/flink/app/application.go

@@ -62,8 +62,8 @@ func StartApp(gatewayClient ccloudv2.GatewayClientInterface, tokenRefreshFunc fu
 	dataStore := store.NewStore(gatewayClient, appController.ExitApplication, &appOptions, synchronizedTokenRefreshFunc)
 	resultFetcher := results.NewResultFetcher(dataStore)
 
-	// Instantiate lsp
-	lspClient := lsp.NewLSPClientWS(getAuthToken, appOptions.GetLSPBaseUrl(), appOptions.GetOrganizationId(), appOptions.GetEnvironmentId())
+	// Instantiate LSP
+	lspClient := lsp.NewWebsocketClient(getAuthToken, appOptions.GetLSPBaseUrl(), appOptions.GetOrganizationId(), appOptions.GetEnvironmentId())
 
 	stdinBefore := utils.GetStdin()
 	consoleParser, err := utils.GetConsoleParser()

pkg/flink/lsp/lsp_completer_ws.go

@@ -80,7 +80,7 @@ func (w *WebsocketLSPClient) refreshWebsocketConnection() {
 	}
 }
 
-func NewLSPClientWS(getAuthToken func() string, baseUrl, organizationId, environmentId string) LspInterface {
+func NewWebsocketClient(getAuthToken func() string, baseUrl, organizationId, environmentId string) LspInterface {
 	lspClient, conn, err := newLSPConnection(baseUrl, getAuthToken(), organizationId, environmentId)
 	if err != nil {
 		return nil

pkg/flink/types/application_options.go

@@ -21,7 +21,7 @@ type ApplicationOptions struct {
 	ServiceAccountId string
 	Verbose          bool
 	LSPBaseUrl       string
-	GatewayURL       string
+	GatewayUrl       string
 	Context          *config.Context
 }
 
@@ -104,7 +104,7 @@ func (a *ApplicationOptions) Validate() error {
 	if a.GetComputePoolId() == "" {
 		missingOptions = append(missingOptions, "ComputePoolId")
 	}
-	if a.GetGatewayURL() == "" {
+	if a.GetGatewayUrl() == "" {
 		missingOptions = append(missingOptions, "GatewayURL")
 	}
 	if len(missingOptions) > 0 {
@@ -190,9 +190,9 @@ func (a *ApplicationOptions) GetLSPBaseUrl() string {
 	return ""
 }
 
-func (a *ApplicationOptions) GetGatewayURL() string {
+func (a *ApplicationOptions) GetGatewayUrl() string {
 	if a != nil {
-		return a.GatewayURL
+		return a.GatewayUrl
 	}
 	return ""
 }

pkg/flink/types/application_options_test.go

@@ -40,7 +40,7 @@ func TestParseApplicationOptionsFromSlices(t *testing.T) {
 		},
 		{
 			name:         "TestParseAppOptionsSuccessfully",
-			configKeys:   []string{"UnsafeTrace", "UserAgent", "EnvironmentId", "EnvironmentName", "OrganizationId", "Database", "ComputePoolId", "ServiceAccountId", "Verbose", "LSPBaseUrl", "GatewayURL"},
+			configKeys:   []string{"UnsafeTrace", "UserAgent", "EnvironmentId", "EnvironmentName", "OrganizationId", "Database", "ComputePoolId", "ServiceAccountId", "Verbose", "LSPBaseUrl", "GatewayUrl"},
 			configValues: []string{"true", "test", "env-123", "test-env", "org-123", "test-database", "lfcp-123", "sa-123", "true", "localhost:8080", "localhost:8000"},
 			expectedAppOptions: &ApplicationOptions{
 				UnsafeTrace:      true,
@@ -53,7 +53,7 @@ func TestParseApplicationOptionsFromSlices(t *testing.T) {
 				ServiceAccountId: "sa-123",
 				Verbose:          true,
 				LSPBaseUrl:       "localhost:8080",
-				GatewayURL:       "localhost:8000",
+				GatewayUrl:       "localhost:8000",
 			},
 			expectedError: false,
 		},