diff --git a/pkg/flink/app/application.go b/pkg/flink/app/application.go
index 8040eced71..5a430ec7d3 100644
--- a/pkg/flink/app/application.go
+++ b/pkg/flink/app/application.go
@@ -88,10 +88,11 @@ func StartApp(gatewayClient ccloudv2.GatewayClientInterface, tokenRefreshFunc fu
 	// Instantiate Component Controllers
 	lspCompleter := lsp.LspCompleter(lspClient, func() lsp.CliContext {
 		return lsp.CliContext{
-			AuthToken:     getAuthToken(),
-			Catalog:       userProperties.Get(config.KeyCatalog),
-			Database:      userProperties.Get(config.KeyDatabase),
-			ComputePoolId: appOptions.GetComputePoolId(),
+			AuthToken:           getAuthToken(),
+			Catalog:             userProperties.Get(config.KeyCatalog),
+			Database:            userProperties.Get(config.KeyDatabase),
+			ComputePoolId:       appOptions.GetComputePoolId(),
+			StatementProperties: userProperties.GetMaskedNonLocalProperties(),
 		}
 	})
 
diff --git a/pkg/flink/internal/controller/input_controller.go b/pkg/flink/internal/controller/input_controller.go
index e40e8dd593..1cce5f8acb 100644
--- a/pkg/flink/internal/controller/input_controller.go
+++ b/pkg/flink/internal/controller/input_controller.go
@@ -141,6 +141,8 @@ func (c *InputController) initPrompt() (prompt.IPrompt, error) {
 		prompt.OptionSelectedSuggestionBGColor(prompt.LightGray),
 		prompt.OptionSuggestionBGColor(prompt.DarkGray),
 		prompt.OptionSetLexer(highlighting.Lexer),
+		prompt.OptionDiagnosticsDetailsBGColor(prompt.Red),
+		prompt.OptionDiagnosticsDetailsTextColor(prompt.White),
 		prompt.OptionSetStatementTerminator(func(lastKeyStroke prompt.Key, buffer *prompt.Buffer) bool {
 			text := buffer.Text()
 			text = strings.TrimSpace(text)
diff --git a/pkg/flink/internal/store/user_properties.go b/pkg/flink/internal/store/user_properties.go
index 832de460ee..285e5c7c34 100644
--- a/pkg/flink/internal/store/user_properties.go
+++ b/pkg/flink/internal/store/user_properties.go
@@ -98,6 +98,19 @@ func (p *UserProperties) GetNonLocalProperties() map[string]string {
 	return nonLocalProperties
 }
 
+// GetMaskedNonLocalProperties returns the same as GetNonLocalProperties but with sensitive values masked
+func (p *UserProperties) GetMaskedNonLocalProperties() map[string]string {
+	maskedProperties := p.GetNonLocalProperties()
+	for key := range maskedProperties {
+		if !strings.HasPrefix(key, config.NamespaceClient) {
+			if hasSensitiveKey(key) {
+				maskedProperties[key] = "hidden"
+			}
+		}
+	}
+	return maskedProperties
+}
+
 func (p *UserProperties) Delete(key string) {
 	defaultValue, isDefaultKey := p.defaultProperties[key]
 	if isDefaultKey {
diff --git a/pkg/flink/internal/store/user_properties_test.go b/pkg/flink/internal/store/user_properties_test.go
index 8425660ea1..22d263fbbe 100644
--- a/pkg/flink/internal/store/user_properties_test.go
+++ b/pkg/flink/internal/store/user_properties_test.go
@@ -174,7 +174,26 @@ func (s *UserPropertiesTestSuite) TestShouldOnlyReturnNonLocalNamespacePropertie
 	s.userProperties.Set(config.KeyCatalog, "test-catalog")
 
 	require.Equal(s.T(), map[string]string{
-		config.KeyCatalog: "test-catalog",
 		"default-key":     "default-value",
+		config.KeyCatalog: "test-catalog",
 	}, s.userProperties.GetNonLocalProperties())
 }
+
+func (s *UserPropertiesTestSuite) TestShouldReturnMaskedProperties() {
+	// Test if non local non sensitive keys are preserved
+	s.userProperties.Set(config.KeyResultsTimeout, "1000")
+	s.userProperties.Set(config.KeyCatalog, "test-catalog")
+
+	expectedProperties := map[string]string{
+		"default-key":     "default-value",
+		config.KeyCatalog: "test-catalog",
+	}
+	require.Equal(s.T(), expectedProperties, s.userProperties.GetMaskedNonLocalProperties())
+
+	// Test if sensitive keys are masked
+	secretKey := config.KeySqlSecrets + "api_key_1"
+	s.userProperties.Set(secretKey, "my-apy-key-value")
+
+	expectedProperties[secretKey] = "hidden"
+	require.Equal(s.T(), expectedProperties, s.userProperties.GetMaskedNonLocalProperties())
+}
diff --git a/pkg/flink/lsp/lsp.go b/pkg/flink/lsp/lsp.go
index 6ad01a344a..58fef4684e 100644
--- a/pkg/flink/lsp/lsp.go
+++ b/pkg/flink/lsp/lsp.go
@@ -26,10 +26,11 @@ type LSPClient struct {
 }
 
 type CliContext struct {
-	AuthToken     string
-	Catalog       string
-	Database      string
-	ComputePoolId string
+	AuthToken           string
+	Catalog             string
+	Database            string
+	ComputePoolId       string
+	StatementProperties map[string]string
 }
 
 func NewLSPClient(conn types.JSONRpcConn) *LSPClient {
diff --git a/pkg/flink/types/user_properties.go b/pkg/flink/types/user_properties.go
index 43de7f24d4..286279eed3 100644
--- a/pkg/flink/types/user_properties.go
+++ b/pkg/flink/types/user_properties.go
@@ -6,6 +6,7 @@ type UserPropertiesInterface interface {
 	Clear()
 	Delete(key string)
 	Get(key string) string
+	GetMaskedNonLocalProperties() map[string]string
 	GetNonLocalProperties() map[string]string
 	GetOrDefault(key string, defaultValue string) string
 	GetOutputFormat() config.OutputFormat