CF-1871 : Add Catalog Update Command

internal/flink/command_catalog.go

@@ -1,11 +1,19 @@
 package flink
 
 import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+
 	"github.com/spf13/cobra"
+	"gopkg.in/yaml.v3"
 
 	cmfsdk "github.com/confluentinc/cmf-sdk-go/v1"
 
 	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/errors"
+	"github.com/confluentinc/cli/v4/pkg/output"
 )
 
 type catalogOut struct {
@@ -25,10 +33,67 @@ func (c *command) newCatalogCommand() *cobra.Command {
 	cmd.AddCommand(c.newCatalogDeleteCommand())
 	cmd.AddCommand(c.newCatalogDescribeCommand())
 	cmd.AddCommand(c.newCatalogListCommand())
+	cmd.AddCommand(c.newCatalogUpdateCommand())
 
 	return cmd
 }
 
+func printCatalogOutput(cmd *cobra.Command, sdkOutputCatalog cmfsdk.KafkaCatalog) error {
+	if output.GetFormat(cmd) == output.Human {
+		table := output.NewTable(cmd)
+		databases := make([]string, 0, len(sdkOutputCatalog.Spec.GetKafkaClusters()))
+		for _, kafkaCluster := range sdkOutputCatalog.Spec.GetKafkaClusters() {
+			databases = append(databases, kafkaCluster.DatabaseName)
+		}
+		var creationTime string
+		if sdkOutputCatalog.GetMetadata().CreationTimestamp != nil {
+			creationTime = *sdkOutputCatalog.GetMetadata().CreationTimestamp
+		}
+		table.Add(&catalogOut{
+			CreationTime: creationTime,
+			Name:         sdkOutputCatalog.GetMetadata().Name,
+			Databases:    databases,
+		})
+		return table.Print()
+	}
+
+	localCatalog := convertSdkCatalogToLocalCatalog(sdkOutputCatalog)
+	return output.SerializedOutput(cmd, localCatalog)
+}
+
+func readCatalogResourceFile(resourceFilePath string) (cmfsdk.KafkaCatalog, error) {
+	data, err := os.ReadFile(resourceFilePath)
+	if err != nil {
+		return cmfsdk.KafkaCatalog{}, fmt.Errorf("failed to read file: %w", err)
+	}
+
+	var genericData map[string]interface{}
+	ext := filepath.Ext(resourceFilePath)
+	switch ext {
+	case ".json":
+		err = json.Unmarshal(data, &genericData)
+	case ".yaml", ".yml":
+		err = yaml.Unmarshal(data, &genericData)
+	default:
+		return cmfsdk.KafkaCatalog{}, errors.NewErrorWithSuggestions(fmt.Sprintf("unsupported file format: %s", ext), "Supported file formats are .json, .yaml, and .yml.")
+	}
+	if err != nil {
+		return cmfsdk.KafkaCatalog{}, fmt.Errorf("failed to parse input file: %w", err)
+	}
+
+	jsonBytes, err := json.Marshal(genericData)
+	if err != nil {
+		return cmfsdk.KafkaCatalog{}, fmt.Errorf("failed to marshal intermediate data: %w", err)
+	}
+
+	var sdkCatalog cmfsdk.KafkaCatalog
+	if err = json.Unmarshal(jsonBytes, &sdkCatalog); err != nil {
+		return cmfsdk.KafkaCatalog{}, fmt.Errorf("failed to bind data to KafkaCatalog model: %w", err)
+	}
+
+	return sdkCatalog, nil
+}
+
 func convertSdkCatalogToLocalCatalog(sdkOutputCatalog cmfsdk.KafkaCatalog) LocalKafkaCatalog {
 	localClusters := make([]LocalKafkaCatalogSpecKafkaClusters, 0, len(sdkOutputCatalog.Spec.GetKafkaClusters()))
 	for _, sdkCluster := range sdkOutputCatalog.Spec.GetKafkaClusters() {

internal/flink/command_catalog_update.go

@@ -0,0 +1,54 @@
+package flink
+
+import (
+	"fmt"
+
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+)
+
+func (c *command) newCatalogUpdateCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "update <resourceFilePath>",
+		Short: "Update a Flink catalog.",
+		Long:  "Update an existing Kafka Catalog in Confluent Platform from a resource file.",
+		Args:  cobra.ExactArgs(1),
+		RunE:  c.catalogUpdate,
+	}
+
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	return cmd
+}
+
+func (c *command) catalogUpdate(cmd *cobra.Command, args []string) error {
+	resourceFilePath := args[0]
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	sdkCatalog, err := readCatalogResourceFile(resourceFilePath)
+	if err != nil {
+		return err
+	}
+
+	catalogName := sdkCatalog.Metadata.Name
+	if catalogName == "" {
+		return fmt.Errorf("catalog name is required: ensure the resource file contains a non-empty \"metadata.name\" field")
+	}
+
+	if err := client.UpdateCatalog(c.createContext(), catalogName, sdkCatalog); err != nil {
+		return err
+	}
+
+	sdkOutputCatalog, err := client.DescribeCatalog(c.createContext(), catalogName)
+	if err != nil {
+		return fmt.Errorf("catalog %q was updated successfully, but failed to retrieve updated details: %w", catalogName, err)
+	}
+
+	return printCatalogOutput(cmd, sdkOutputCatalog)
+}

pkg/flink/cmf_rest_client.go

@@ -565,6 +565,14 @@ func (cmfClient *CmfRestClient) ListCatalog(ctx context.Context) ([]cmfsdk.Kafka
 	return catalogs, nil
 }
 
+func (cmfClient *CmfRestClient) UpdateCatalog(ctx context.Context, catalogName string, kafkaCatalog cmfsdk.KafkaCatalog) error {
+	httpResponse, err := cmfClient.SQLApi.UpdateKafkaCatalog(ctx, catalogName).KafkaCatalog(kafkaCatalog).Execute()
+	if parsedErr := parseSdkError(httpResponse, err); parsedErr != nil {
+		return fmt.Errorf(`failed to update Kafka Catalog "%s": %s`, catalogName, parsedErr)
+	}
+	return nil
+}
+
 func (cmfClient *CmfRestClient) DeleteCatalog(ctx context.Context, catalogName string) error {
 	httpResp, err := cmfClient.SQLApi.DeleteKafkaCatalog(ctx, catalogName).Execute()
 	return parseSdkError(httpResp, err)

test/fixtures/input/flink/catalog/update-invalid-failure.json

@@ -0,0 +1,17 @@
+{
+  "apiVersion": "cmf/api/v1/catalog",
+  "kind": "KafkaCatalog",
+  "metadata": {
+    "name": "invalid-catalog"
+  },
+  "spec": {
+    "kafkaClusters": [
+      {
+        "databaseName": "dev"
+      },
+      {
+        "databaseName": "prod"
+      }
+    ]
+  }
+}

test/fixtures/input/flink/catalog/update-invalid-failure.yaml

@@ -0,0 +1,8 @@
+apiVersion: cmf/api/v1/catalog
+kind: KafkaCatalog
+metadata:
+  name: invalid-catalog
+spec:
+  kafkaClusters:
+    - databaseName: dev
+    - databaseName: prod

test/fixtures/input/flink/catalog/update-successful.json

@@ -0,0 +1,17 @@
+{
+  "apiVersion": "cmf/api/v1/catalog",
+  "kind": "KafkaCatalog",
+  "metadata": {
+    "name": "test-catalog"
+  },
+  "spec": {
+    "kafkaClusters": [
+      {
+        "databaseName": "dev"
+      },
+      {
+        "databaseName": "prod"
+      }
+    ]
+  }
+}

test/fixtures/input/flink/catalog/update-successful.yaml

@@ -0,0 +1,8 @@
+apiVersion: cmf/api/v1/catalog
+kind: KafkaCatalog
+metadata:
+  name: test-catalog
+spec:
+  kafkaClusters:
+    - databaseName: dev
+    - databaseName: prod

test/fixtures/output/flink/catalog/help-onprem.golden

@@ -8,6 +8,7 @@ Available Commands:
   delete      Delete one or more Flink catalogs in Confluent Platform.
   describe    Describe a Flink catalog in Confluent Platform.
   list        List Flink catalogs in Confluent Platform.
+  update      Update a Flink catalog.
 
 Global Flags:
   -h, --help            Show help for this command.

test/fixtures/output/flink/catalog/update-help-onprem.golden

@@ -0,0 +1,16 @@
+Update an existing Kafka Catalog in Confluent Platform from a resource file.
+
+Usage:
+  confluent flink catalog update <resourceFilePath> [flags]
+
+Flags:
+      --url string                          Base URL of the Confluent Manager for Apache Flink (CMF). Environment variable "CONFLUENT_CMF_URL" may be set in place of this flag.
+      --client-key-path string              Path to client private key for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_KEY_PATH" may be set in place of this flag.
+      --client-cert-path string             Path to client cert to be verified by Confluent Manager for Apache Flink. Include for mTLS authentication. Environment variable "CONFLUENT_CMF_CLIENT_CERT_PATH" may be set in place of this flag.
+      --certificate-authority-path string   Path to a PEM-encoded Certificate Authority to verify the Confluent Manager for Apache Flink connection. Environment variable "CONFLUENT_CMF_CERTIFICATE_AUTHORITY_PATH" may be set in place of this flag.
+  -o, --output string                       Specify the output format as "human", "json", or "yaml". (default "human")
+
+Global Flags:
+  -h, --help            Show help for this command.
+      --unsafe-trace    Equivalent to -vvvv, but also log HTTP requests and responses which might contain plaintext secrets.
+  -v, --verbose count   Increase verbosity (-v for warn, -vv for info, -vvv for debug, -vvvv for trace).

test/fixtures/output/flink/catalog/update-invalid-failure.golden

@@ -0,0 +1 @@
+Error: failed to update Kafka Catalog "invalid-catalog": The catalog name is invalid

test/fixtures/output/flink/catalog/update-success-json.golden

@@ -0,0 +1,23 @@
+{
+  "apiVersion": "",
+  "kind": "",
+  "metadata": {
+    "name": "test-catalog",
+    "creationTimestamp": "2025-08-05 12:00:00 +0000 UTC"
+  },
+  "spec": {
+    "srInstance": {
+      "connectionConfig": null
+    },
+    "kafkaClusters": [
+      {
+        "databaseName": "test-database",
+        "connectionConfig": null
+      },
+      {
+        "databaseName": "test-database-2",
+        "connectionConfig": null
+      }
+    ]
+  }
+}

test/fixtures/output/flink/catalog/update-success-yaml.golden

@@ -0,0 +1,13 @@
+apiVersion: ""
+kind: ""
+metadata:
+    name: test-catalog
+    creationTimestamp: 2025-08-05 12:00:00 +0000 UTC
+spec:
+    srInstance:
+        connectionConfig: {}
+    kafkaClusters:
+        - databaseName: test-database
+          connectionConfig: {}
+        - databaseName: test-database-2
+          connectionConfig: {}

test/fixtures/output/flink/catalog/update-success.golden

@@ -0,0 +1,5 @@
++---------------+--------------------------------+
+| Creation Time | 2025-08-05 12:00:00 +0000 UTC  |
+| Name          | test-catalog                   |
+| Databases     | test-database, test-database-2 |
++---------------+--------------------------------+

test/flink_onprem_test.go

@@ -410,6 +410,19 @@
 	runIntegrationTestsWithMultipleAuth(s, tests)
 }
 
+func (s *CLITestSuite) TestFlinkCatalogUpdateOnPrem() {
+	tests := []CLITest{
+		// success
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json", fixture: "flink/catalog/update-success.golden"},
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json --output json", fixture: "flink/catalog/update-success-json.golden"},
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json --output yaml", fixture: "flink/catalog/update-success-yaml.golden"},
+		// failure
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-invalid-failure.json", fixture: "flink/catalog/update-invalid-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
 func (s *CLITestSuite) TestFlinkStatementCreateOnPrem() {
 	tests := []CLITest{
 		// success
@@ -607,6 +620,25 @@
 	runIntegrationTestsWithMultipleAuth(s, tests)
 }
 
+func (s *CLITestSuite) TestFlinkCatalogUpdateWithYAML() {
+	tests := []CLITest{
+		// success scenarios with JSON files
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json", fixture: "flink/catalog/update-success.golden"},
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json --output json", fixture: "flink/catalog/update-success-json.golden"},
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json --output yaml", fixture: "flink/catalog/update-success-yaml.golden"},
+		// failure scenarios with JSON files
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-invalid-failure.json", fixture: "flink/catalog/update-invalid-failure.golden", exitCode: 1},
+		// YAML file tests
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.yaml", fixture: "flink/catalog/update-success.golden"},
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.yaml --output json", fixture: "flink/catalog/update-success-json.golden"},
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.yaml --output yaml", fixture: "flink/catalog/update-success-yaml.golden"},
+		// YAML file failure scenarios
+		{args: "flink catalog update test/fixtures/input/flink/catalog/update-invalid-failure.yaml", fixture: "flink/catalog/update-invalid-failure.golden", exitCode: 1},
+	}
+
+	runIntegrationTestsWithMultipleAuth(s, tests)
+}
+
 func (s *CLITestSuite) TestFlinkShellOnPrem() {
 	tests := []flinkShellTest{
 		{

test/test-server/flink_onprem_handler.go

@@ -897,7 +897,7 @@
 
 			catName := catalog.GetMetadata().Name
 
 			if catName == "invalid-catalog" {
 				http.Error(w, "The Kafka catalog object from resource file is invalid", http.StatusUnprocessableEntity)
 				return
 			}
@@ -918,7 +918,7 @@
 }
 
 // Handler for "cmf/api/v1/catalogs/kafka/{catName}"
-// Used by describe, delete catalog, no update catalog.
+// Used by describe, update, delete catalog.
 func handleCmfCatalog(t *testing.T) http.HandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request) {
 		handleLoginType(t, r)
@@ -937,6 +937,18 @@
 			err := json.NewEncoder(w).Encode(catalog)
 			require.NoError(t, err)
 			return
+		case http.MethodPut:
+			if catalogName == "invalid-catalog" {
+				http.Error(w, "The catalog name is invalid", http.StatusNotFound)
+				return
+			}
+
+			req := new(cmfsdk.KafkaCatalog)
+			err := json.NewDecoder(r.Body).Decode(req)
+			require.NoError(t, err)
+
+			w.WriteHeader(http.StatusOK)
+			return
 		case http.MethodDelete:
 			if catalogName == "non-exist-catalog" {
 				http.Error(w, "", http.StatusNotFound)