Skip to content

CF-1871 : Add Catalog Update Command #3304

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Paras Negi (paras-negi-flink) wants to merge 2 commits into
base: main
Choose a base branch
from
Open

CF-1871 : Add Catalog Update Command #3304

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
6571fb1
Add Catalog Update Command
paras-negi-flink Apr 3, 2026
f712e73
[CF-1871] Address review comments
paras-negi-flink May 18, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
65 changes: 65 additions & 0 deletions internal/flink/command_catalog.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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 {
Expand All @@ -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)
}
Comment on lines +41 to +62
Copy link

Copilot AI Apr 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

printCatalogOutput centralizes catalog rendering, but catalogCreate and catalogDescribe still contain their own (nearly identical) output formatting logic. To avoid future drift between commands, consider switching those commands to call printCatalogOutput as well.

Copilot uses AI. Check for mistakes.
Copy link
Copy Markdown
Contributor Author

@paras-negi-flink Paras Negi (paras-negi-flink) Apr 3, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Robert Metzger (@rmetzger) Kumar Mallikarjuna (@kumar-mallikarjuna) does it make sense to touch other existing files to reduce duplication, or should we tackle that in another change ?

Copy link
Copy Markdown

@rmetzger Robert Metzger (rmetzger) Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think it makes sense to fix this as part of this PR

Copy link
Copy Markdown
Contributor

@channingdong Channing Dong (channingdong) Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Agree with Copilot and Robert Metzger (@rmetzger) 's comment, maybe we can apply this idea to the catalog database as well.

Copy link
Copy Markdown
Contributor Author

@paras-negi-flink Paras Negi (paras-negi-flink) May 18, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done — catalogCreate and catalogDescribe now both call printCatalogOutput.


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
}
Comment thread
paras-negi-flink marked this conversation as resolved.

func convertSdkCatalogToLocalCatalog(sdkOutputCatalog cmfsdk.KafkaCatalog) LocalKafkaCatalog {
localClusters := make([]LocalKafkaCatalogSpecKafkaClusters, 0, len(sdkOutputCatalog.Spec.GetKafkaClusters()))
for _, sdkCluster := range sdkOutputCatalog.Spec.GetKafkaClusters() {
Expand Down
54 changes: 54 additions & 0 deletions internal/flink/command_catalog_update.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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.",
Comment thread
paras-negi-flink marked this conversation as resolved.
Outdated
Long: "Update an existing Kafka Catalog in Confluent Platform from a resource file.",
Copy link
Copy Markdown
Contributor

@channingdong Channing Dong (channingdong) Apr 14, 2026 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We want to keep it consistent.

Suggested change
Long: "Update an existing Kafka Catalog in Confluent Platform from a resource file.",
Long: "Update an existing Flink catalog in Confluent Platform from a resource file.",

Copy link
Copy Markdown
Contributor

@channingdong Channing Dong (channingdong) Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

And by the way, is there a convention for the user to know which fields are mutable from an update operation?

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 {
Copy link
Copy Markdown
Contributor

@channingdong Channing Dong (channingdong) Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Same comment as the catalog database, is it the design that Update command doesn't return the updated catalog to make this an atomic operation? I am seeing an additional describe call below.

Copy link
Copy Markdown
Contributor Author

@paras-negi-flink Paras Negi (paras-negi-flink) May 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes thats correct. UpdateKafkaCatalogExecute returns (*http.Response, error) only — no body. The follow-up DescribeCatalog is the only way to render the updated resource until CMF changes the PUT endpoint to return the catalog.

As discussed, we will address this in a follow up PR.

return err
}

sdkOutputCatalog, err := client.DescribeCatalog(c.createContext(), catalogName)
Comment thread
paras-negi-flink marked this conversation as resolved.
if err != nil {
return fmt.Errorf("catalog %q was updated successfully, but failed to retrieve updated details: %w", catalogName, err)
}

return printCatalogOutput(cmd, sdkOutputCatalog)
Copy link
Copy Markdown
Contributor

@channingdong Channing Dong (channingdong) Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's use this unified print out function for the create command as well.

Copy link
Copy Markdown
Contributor Author

@paras-negi-flink Paras Negi (paras-negi-flink) May 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done. catalogCreate now ends with return printCatalogOutput(cmd, sdkOutputCatalog)

}
8 changes: 8 additions & 0 deletions pkg/flink/cmf_rest_client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
17 changes: 17 additions & 0 deletions test/fixtures/input/flink/catalog/update-invalid-failure.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
{
"apiVersion": "cmf/api/v1/catalog",
"kind": "KafkaCatalog",
"metadata": {
"name": "invalid-catalog"
},
"spec": {
"kafkaClusters": [
{
"databaseName": "dev"
},
{
"databaseName": "prod"
}
]
}
}
8 changes: 8 additions & 0 deletions test/fixtures/input/flink/catalog/update-invalid-failure.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
apiVersion: cmf/api/v1/catalog
kind: KafkaCatalog
metadata:
name: invalid-catalog
spec:
kafkaClusters:
- databaseName: dev
- databaseName: prod
17 changes: 17 additions & 0 deletions test/fixtures/input/flink/catalog/update-successful.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,17 @@
{
"apiVersion": "cmf/api/v1/catalog",
"kind": "KafkaCatalog",
"metadata": {
"name": "test-catalog"
},
"spec": {
"kafkaClusters": [
{
"databaseName": "dev"
},
{
"databaseName": "prod"
}
]
}
}
8 changes: 8 additions & 0 deletions test/fixtures/input/flink/catalog/update-successful.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
apiVersion: cmf/api/v1/catalog
kind: KafkaCatalog
metadata:
name: test-catalog
spec:
kafkaClusters:
- databaseName: dev
- databaseName: prod
1 change: 1 addition & 0 deletions test/fixtures/output/flink/catalog/help-onprem.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
16 changes: 16 additions & 0 deletions test/fixtures/output/flink/catalog/update-help-onprem.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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).
1 change: 1 addition & 0 deletions test/fixtures/output/flink/catalog/update-invalid-failure.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
Error: failed to update Kafka Catalog "invalid-catalog": The catalog name is invalid
23 changes: 23 additions & 0 deletions test/fixtures/output/flink/catalog/update-success-json.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,23 @@
{
"apiVersion": "",
"kind": "",
Copy link
Copy Markdown

@rmetzger Robert Metzger (rmetzger) Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why are apiVersion and kind empty?

Copy link
Copy Markdown
Contributor Author

@paras-negi-flink Paras Negi (paras-negi-flink) May 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed.

"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
}
Copy link
Copy Markdown
Contributor

@channingdong Channing Dong (channingdong) Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

question: Are the connectionConfig designed to be null or empty map?

Copy link
Copy Markdown
Contributor Author

@paras-negi-flink Paras Negi (paras-negi-flink) May 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

No, it should always be a populated map in CMF's response. Fixed now

]
}
}
13 changes: 13 additions & 0 deletions test/fixtures/output/flink/catalog/update-success-yaml.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
apiVersion: ""
kind: ""
Copy link
Copy Markdown

@rmetzger Robert Metzger (rmetzger) Apr 7, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

this seems off

Copy link
Copy Markdown
Contributor Author

@paras-negi-flink Paras Negi (paras-negi-flink) May 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed.

Copy link
Copy Markdown
Contributor

@channingdong Channing Dong (channingdong) Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Please fill these fields, and similar golden files.

Copy link
Copy Markdown
Contributor Author

@paras-negi-flink Paras Negi (paras-negi-flink) May 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed.

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: {}
5 changes: 5 additions & 0 deletions test/fixtures/output/flink/catalog/update-success.golden
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
+---------------+--------------------------------+
| Creation Time | 2025-08-05 12:00:00 +0000 UTC |
| Name | test-catalog |
| Databases | test-database, test-database-2 |
+---------------+--------------------------------+
32 changes: 32 additions & 0 deletions test/flink_onprem_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"},

Check failure on line 416 in test/flink_onprem_test.go

View check run for this annotation

SonarQube-Confluent / SonarQube Code Analysis

Define a constant instead of duplicating this literal "flink/catalog/update-success.golden" 3 times. 

[S1192] String literals should not be duplicated
 See more on https://sonarqube.confluent.io/project/issues?id=cli&pullRequest=3304&issues=cb2e5dd0-f916-4786-8961-10cdf062da42&open=cb2e5dd0-f916-4786-8961-10cdf062da42
{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json --output json", fixture: "flink/catalog/update-success-json.golden"},

Check failure on line 417 in test/flink_onprem_test.go

View check run for this annotation

SonarQube-Confluent / SonarQube Code Analysis

Define a constant instead of duplicating this literal "flink/catalog/update-success-json.golden" 3 times. 

[S1192] String literals should not be duplicated
 See more on https://sonarqube.confluent.io/project/issues?id=cli&pullRequest=3304&issues=1c9fc779-f6fb-4b49-9915-3e96cd60e40d&open=1c9fc779-f6fb-4b49-9915-3e96cd60e40d
{args: "flink catalog update test/fixtures/input/flink/catalog/update-successful.json --output yaml", fixture: "flink/catalog/update-success-yaml.golden"},

Check failure on line 418 in test/flink_onprem_test.go

View check run for this annotation

SonarQube-Confluent / SonarQube Code Analysis

Define a constant instead of duplicating this literal "flink/catalog/update-success-yaml.golden" 3 times. 

[S1192] String literals should not be duplicated
 See more on https://sonarqube.confluent.io/project/issues?id=cli&pullRequest=3304&issues=98fbdc51-b110-4929-b457-a54def18c068&open=98fbdc51-b110-4929-b457-a54def18c068
// failure
{args: "flink catalog update test/fixtures/input/flink/catalog/update-invalid-failure.json", fixture: "flink/catalog/update-invalid-failure.golden", exitCode: 1},

Check failure on line 420 in test/flink_onprem_test.go

View check run for this annotation

SonarQube-Confluent / SonarQube Code Analysis

Define a constant instead of duplicating this literal "flink/catalog/update-invalid-failure.golden" 3 times. 

[S1192] String literals should not be duplicated
 See more on https://sonarqube.confluent.io/project/issues?id=cli&pullRequest=3304&issues=3149a87a-192b-46e1-a3ab-88b3a1ca8371&open=3149a87a-192b-46e1-a3ab-88b3a1ca8371
}

runIntegrationTestsWithMultipleAuth(s, tests)
}

func (s *CLITestSuite) TestFlinkStatementCreateOnPrem() {
tests := []CLITest{
// success
Expand Down Expand Up @@ -607,6 +620,25 @@
runIntegrationTestsWithMultipleAuth(s, tests)
}

func (s *CLITestSuite) TestFlinkCatalogUpdateWithYAML() {
Comment thread
paras-negi-flink marked this conversation as resolved.
Outdated
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},
Copy link
Copy Markdown
Contributor

@channingdong Channing Dong (channingdong) Apr 14, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should these be json format files? Or are these totally redundant?

Copy link
Copy Markdown
Contributor Author

@paras-negi-flink Paras Negi (paras-negi-flink) May 18, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Dropped them and renamed the function to TestFlinkCatalogUpdateOnPremWithYAML, so it's now YAML-only.

// 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},
}
Comment thread
paras-negi-flink marked this conversation as resolved.
Outdated

runIntegrationTestsWithMultipleAuth(s, tests)
}

func (s *CLITestSuite) TestFlinkShellOnPrem() {
tests := []flinkShellTest{
{
Expand Down
14 changes: 13 additions & 1 deletion test/test-server/flink_onprem_handler.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -897,7 +897,7 @@

catName := catalog.GetMetadata().Name

if catName == "invalid-catalog" {

Check failure on line 900 in test/test-server/flink_onprem_handler.go

View check run for this annotation

SonarQube-Confluent / SonarQube Code Analysis

Define a constant instead of duplicating this literal "invalid-catalog" 3 times. 

[S1192] String literals should not be duplicated
 See more on https://sonarqube.confluent.io/project/issues?id=cli&pullRequest=3304&issues=2dffbeb0-9b33-45c0-869f-a7489170860a&open=2dffbeb0-9b33-45c0-869f-a7489170860a
http.Error(w, "The Kafka catalog object from resource file is invalid", http.StatusUnprocessableEntity)
return
}
Expand All @@ -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)
Expand All @@ -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)
Expand Down