Skip to content

Update USM commands to remove on-prem to cloud Kafka ID mapping and simplify connect operations #3416

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
Karan Singh Khati (karanskhati) wants to merge 2 commits into
base: main
Choose a base branch
from
Open

Update USM commands to remove on-prem to cloud Kafka ID mapping and simplify connect operations #3416

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
5217292
Update USM commands to remove on-prem to cloud Kafka ID mapping and s…
karanskhati Jul 28, 2026
e141957
Add UsmKafkaClusterId to connect cluster handlers for improved identi…
karanskhati Jul 28, 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
2 changes: 1 addition & 1 deletion go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -52,7 +52,7 @@ require (
github.com/confluentinc/ccloud-sdk-go-v2/srcm v0.7.3
github.com/confluentinc/ccloud-sdk-go-v2/sso v0.0.1
github.com/confluentinc/ccloud-sdk-go-v2/tableflow v0.6.0
github.com/confluentinc/ccloud-sdk-go-v2/usm v0.1.0
github.com/confluentinc/ccloud-sdk-go-v2/usm v0.2.0
github.com/confluentinc/cmf-sdk-go v0.0.8
github.com/confluentinc/confluent-kafka-go/v2 v2.14.2
github.com/confluentinc/go-editor v0.11.0
Expand Down
195 changes: 195 additions & 0 deletions go.sum
View file
Open in desktop

Large diffs are not rendered by default.

20 changes: 2 additions & 18 deletions internal/unified-stream-manager/command_connect.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,8 +10,6 @@ import (
"github.com/confluentinc/cli/v4/pkg/output"
)

const kafkaClusterNotFoundErrorMsg = "USM Kafka cluster corresponding to Confluent Platform Kafka cluster %s not found"

type connectOut struct {
Id string `human:"ID" serialized:"id"`
ConfluentPlatformConnectCluster string `human:"Confluent Platform Connect Cluster" serialized:"confluent_platform_connect_cluster"`
Expand Down Expand Up @@ -70,11 +68,11 @@ func (c *command) autocompleteConnectClusters() []string {
return suggestions
}

func printConnectTable(cmd *cobra.Command, connect usmv1.UsmV1ConnectCluster, usmKafkaClusterId string) error {
func printConnectTable(cmd *cobra.Command, connect usmv1.UsmV1ConnectCluster) error {
out := &connectOut{
Id: connect.GetId(),
ConfluentPlatformConnectCluster: connect.GetConfluentPlatformConnectClusterId(),
USMKafkaClusterId: usmKafkaClusterId,
USMKafkaClusterId: connect.GetUsmKafkaClusterId(),
ConfluentPlatformKafkaClusterId: connect.GetKafkaClusterId(),
Cloud: connect.GetCloud(),
Region: connect.GetRegion(),
Expand All @@ -85,17 +83,3 @@ func printConnectTable(cmd *cobra.Command, connect usmv1.UsmV1ConnectCluster, us
table.Add(out)
return table.Print()
}

func (c *command) getOnPremToCloudKafkaIdMap(environment string) (map[string]string, error) {
clusters, err := c.V2Client.ListUsmKafkaClusters(environment)
if err != nil {
return nil, err
}

idMap := make(map[string]string, len(clusters))
for _, cluster := range clusters {
idMap[cluster.GetConfluentPlatformKafkaClusterId()] = cluster.GetId()
}

return idMap, nil
}
13 changes: 1 addition & 12 deletions internal/unified-stream-manager/command_connect_describe.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import (

pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
"github.com/confluentinc/cli/v4/pkg/examples"
"github.com/confluentinc/cli/v4/pkg/log"
)

func (c *command) newConnectDescribeCommand() *cobra.Command {
Expand Down Expand Up @@ -41,15 +40,5 @@ func (c *command) describeConnect(cmd *cobra.Command, args []string) error {
return err
}

onPremToCloudKafkaIdMap, err := c.getOnPremToCloudKafkaIdMap(environmentId)
if err != nil {
return err
}

usmKafkaClusterId, ok := onPremToCloudKafkaIdMap[cluster.GetKafkaClusterId()]
if !ok {
log.CliLogger.Errorf(kafkaClusterNotFoundErrorMsg, cluster.GetKafkaClusterId())
}

return printConnectTable(cmd, cluster, usmKafkaClusterId)
return printConnectTable(cmd, cluster)
}
13 changes: 1 addition & 12 deletions internal/unified-stream-manager/command_connect_list.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,6 @@ import (

pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
"github.com/confluentinc/cli/v4/pkg/examples"
"github.com/confluentinc/cli/v4/pkg/log"
"github.com/confluentinc/cli/v4/pkg/output"
)

Expand Down Expand Up @@ -36,27 +35,17 @@ func (c *command) listConnect(cmd *cobra.Command, args []string) error {
return err
}

onPremToCloudKafkaIdMap, err := c.getOnPremToCloudKafkaIdMap(environmentId)
if err != nil {
return err
}

clusters, err := c.V2Client.ListUsmConnectClusters(environmentId)
if err != nil {
return err
}

list := output.NewList(cmd)
for _, cluster := range clusters {
usmKafkaClusterId, ok := onPremToCloudKafkaIdMap[cluster.GetKafkaClusterId()]
if !ok {
log.CliLogger.Errorf(kafkaClusterNotFoundErrorMsg, cluster.GetKafkaClusterId())
}

out := &connectOut{
Id: cluster.GetId(),
ConfluentPlatformConnectCluster: cluster.GetConfluentPlatformConnectClusterId(),
USMKafkaClusterId: usmKafkaClusterId,
USMKafkaClusterId: cluster.GetUsmKafkaClusterId(),
ConfluentPlatformKafkaClusterId: cluster.GetKafkaClusterId(),
Cloud: cluster.GetCloud(),
Region: cluster.GetRegion(),
Expand Down
13 changes: 1 addition & 12 deletions internal/unified-stream-manager/command_connect_register.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@ import (

pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
"github.com/confluentinc/cli/v4/pkg/examples"
"github.com/confluentinc/cli/v4/pkg/log"
)

func (c *command) newConnectRegisterCommand() *cobra.Command {
Expand Down Expand Up @@ -45,21 +44,11 @@ func (c *command) registerConnect(cmd *cobra.Command, args []string) error {
return err
}

onPremToCloudKafkaIdMap, err := c.getOnPremToCloudKafkaIdMap(environmentId)
if err != nil {
return err
}

kafkaClusterId, err := cmd.Flags().GetString("confluent-platform-kafka-cluster")
if err != nil {
return err
}

usmKafkaClusterId, ok := onPremToCloudKafkaIdMap[kafkaClusterId]
if !ok {
log.CliLogger.Errorf(kafkaClusterNotFoundErrorMsg, kafkaClusterId)
}

connectClusterRequest := usmv1.UsmV1ConnectCluster{
ConfluentPlatformConnectClusterId: usmv1.PtrString(args[0]),
KafkaClusterId: usmv1.PtrString(kafkaClusterId),
Expand Down Expand Up @@ -87,5 +76,5 @@ func (c *command) registerConnect(cmd *cobra.Command, args []string) error {
return err
}

return printConnectTable(cmd, cluster, usmKafkaClusterId)
return printConnectTable(cmd, cluster)
}
4 changes: 4 additions & 0 deletions test/test-server/usm_handlers.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,7 +71,7 @@
err := json.NewDecoder(r.Body).Decode(usmKafkaCluster)
require.NoError(t, err)

usmKafkaCluster.SetId("usmkc-abc123")

Check failure on line 74 in test/test-server/usm_handlers.go

View check run for this annotation

SonarQube-Confluent / SonarQube Code Analysis

Define a constant instead of duplicating this literal "usmkc-abc123" 5 times. 

[S1192] String literals should not be duplicated
 See more on https://sonarqube.confluent.io/project/issues?id=cli&pullRequest=3416&issues=f15b38b5-e9b0-4ffa-ad49-f248d7254e9a&open=f15b38b5-e9b0-4ffa-ad49-f248d7254e9a

err = json.NewEncoder(w).Encode(usmKafkaCluster)
require.NoError(t, err)
Expand Down Expand Up @@ -145,6 +145,7 @@
require.NoError(t, err)

usmConnectCluster.SetId("usmcc-abc123")
usmConnectCluster.SetUsmKafkaClusterId("usmkc-abc123")
if usmConnectCluster.GetCloud() == "" {
usmConnectCluster.SetCloud("AWS")
usmConnectCluster.SetRegion("us-west-2")
Expand All @@ -163,6 +164,7 @@
Id: usmv1.PtrString("usmcc-abc123"),
ConfluentPlatformConnectClusterId: usmv1.PtrString("connect-group-xyz123"),
KafkaClusterId: usmv1.PtrString("4k0R9d1GTS5tI9f4Y2xZ0Q"),
UsmKafkaClusterId: usmv1.PtrString("usmkc-abc123"),
Cloud: usmv1.PtrString("AWS"),
Region: usmv1.PtrString("us-west-2"),
Environment: &usmv1.EnvScopedObjectReference{Id: environmentId},
Expand All @@ -171,6 +173,7 @@
Id: usmv1.PtrString("usmcc-def456"),
ConfluentPlatformConnectClusterId: usmv1.PtrString("connect-group-xyz456"),
KafkaClusterId: usmv1.PtrString("5k0R9d1GTS5tI9f4Y2xZ0Q"),
UsmKafkaClusterId: usmv1.PtrString("usmkc-def456"),
Cloud: usmv1.PtrString("Azure"),
Region: usmv1.PtrString("eastus"),
Environment: &usmv1.EnvScopedObjectReference{Id: environmentId},
Expand Down Expand Up @@ -205,6 +208,7 @@
Id: usmv1.PtrString(id),
ConfluentPlatformConnectClusterId: usmv1.PtrString("connect-group-xyz123"),
KafkaClusterId: usmv1.PtrString("4k0R9d1GTS5tI9f4Y2xZ0Q"),
UsmKafkaClusterId: usmv1.PtrString("usmkc-abc123"),
Cloud: usmv1.PtrString("AWS"),
Region: usmv1.PtrString("us-west-2"),
Environment: &usmv1.EnvScopedObjectReference{Id: environmentId},
Expand Down