Fixes statement name generation for CP Flink (#3242)

Author: vsantwana

internal/flink/command_statement_create_onprem.go

@@ -49,7 +49,7 @@ func (c *command) newStatementCreateCommandOnPrem() *cobra.Command {
 
 func (c *command) statementCreateOnPrem(cmd *cobra.Command, args []string) error {
 	// Flink statement name can be automatically generated or provided by the user
-	name := types.GenerateStatementName()
+	name := types.GenerateStatementNameForOnPrem()
 	if len(args) == 1 {
 		name = args[0]
 	}

pkg/flink/internal/store/store_onprem.go

@@ -60,7 +60,7 @@ func (s *StoreOnPrem) ProcessStatement(statement string) (*types.ProcessedStatem
 		return result, sErr
 	}
 
-	statementName := s.Properties.GetOrDefault(config.KeyStatementName, types.GenerateStatementName())
+	statementName := s.Properties.GetOrDefault(config.KeyStatementName, types.GenerateStatementNameForOnPrem())
 	if len(statementName) > 45 { // on-prem name length limit
 		statementName = statementName[0:45]
 	}

pkg/flink/internal/store/store_test.go

@@ -55,6 +55,13 @@ func (s *StoreTestSuite) TestGenerateStatementName() {
 	}
 }
 
+func (s *StoreTestSuite) TestGenerateStatementNameForOnPrem() {
+	statementRegex := `^cli-\d{8}-\d{6}-[a-f0-9]{24}$`
+	for i := 0; i < 10; i++ {
+		s.Require().Regexp(statementRegex, types.GenerateStatementNameForOnPrem())
+	}
+}
+
 func TestStoreProcessLocalStatement(t *testing.T) {
 	// Create new stores
 	stores := make([]types.StoreInterface, 2)

pkg/flink/types/statement.go

@@ -1,6 +1,8 @@
 package types
 
 import (
+	"crypto/rand"
+	"encoding/hex"
 	"fmt"
 	"time"
 
@@ -14,3 +16,17 @@ func GenerateStatementName() string {
 	id := uuid.New().String()
 	return fmt.Sprintf("%s-%s-%s-%s", clientName, date, localTime, id)
 }
+
+func GenerateStatementNameForOnPrem() string {
+	clientName := "cli"
+	timeNow := time.Now()
+	date := timeNow.Format("20060102")    // 8 chars
+	localTime := timeNow.Format("150405") // 6 chars
+	// 12 random bytes => 24 hex chars
+	b := make([]byte, 12)
+	if _, err := rand.Read(b); err != nil {
+		panic(fmt.Sprintf("unable to generate random bytes for statement name: %v", err))
+	}
+	randomHex := hex.EncodeToString(b)
+	return fmt.Sprintf("%s-%s-%s-%s", clientName, date, localTime, randomHex)
+}