Merge pull request #1021 from redpanda-data/sj/deps/connect-client-up…

…date Sj/deps/connect client update
redpanda-data · Jan 22, 2024 · 7af5e28 · 7af5e28
2 parents bf800d6 + 1dd80c5
commit 7af5e28
Show file tree

Hide file tree

Showing 5 changed files with 78 additions and 229 deletions.
diff --git a/backend/go.mod b/backend/go.mod
@@ -10,7 +10,7 @@ require (
 	github.com/bufbuild/protovalidate-go v0.4.3
 	github.com/carlmjohnson/requests v0.23.5
 	github.com/cloudhut/common v0.10.0
-	github.com/cloudhut/connect-client v0.0.0-20240111195309-ea134bd418d2
+	github.com/cloudhut/connect-client v0.0.0-20240122153328-02a3103805d8
 	github.com/docker/go-connections v0.5.0
 	github.com/dop251/goja v0.0.0-20231027120936-b396bb4c349d
 	github.com/getkin/kin-openapi v0.122.0

diff --git a/backend/go.sum b/backend/go.sum
@@ -79,6 +79,8 @@ github.com/cloudhut/common v0.10.0 h1:Z5nhW3yo60bJSGBJEq6MZurlu3XrYJFMPWPCsVJFcD
 github.com/cloudhut/common v0.10.0/go.mod h1:qbvZg1TgR2mO11rS3d/NCGeUzUzzdEMHTivsSeFjO2A=
 github.com/cloudhut/connect-client v0.0.0-20240111195309-ea134bd418d2 h1:APWoZZP3vqKwV2C4rWmR3FXclcCs0WG0UDCN4NP3Byk=
 github.com/cloudhut/connect-client v0.0.0-20240111195309-ea134bd418d2/go.mod h1:khC9xPwJVKZQ02jtn3n64gEgOKtlVrcWuSOvmj5hSO4=
+github.com/cloudhut/connect-client v0.0.0-20240122153328-02a3103805d8 h1:/vAjE5Gvhj88G2+xziuqhrSRCPjmDeuDakTaBz3s7GI=
+github.com/cloudhut/connect-client v0.0.0-20240122153328-02a3103805d8/go.mod h1:khC9xPwJVKZQ02jtn3n64gEgOKtlVrcWuSOvmj5hSO4=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
 github.com/containerd/containerd v1.7.11 h1:lfGKw3eU35sjV0aG2eYZTiwFEY1pCzxdzicHP3SZILw=

diff --git a/backend/pkg/api/connect/service/kafkaconnect/mapper.go b/backend/pkg/api/connect/service/kafkaconnect/mapper.go
@@ -52,7 +52,7 @@ func (m mapper) connectorsHTTPResponseToProto(httpResponse kafkaconnect.ClusterC
 			Tasks:         m.taskInfoListToProtoStatus(connector.Tasks),
 			Type:          connector.Type,
 			Errors:        errors,
-			HolisticState: m.holisticStateToProto(connector.State),
+			HolisticState: m.holisticStateToProto(connector.Status),
 		}
 	}
 

diff --git a/backend/pkg/connect/get_connectors.go b/backend/pkg/connect/get_connectors.go
@@ -503,28 +503,11 @@ func getHolisticStateFromConnector(status con.ConnectorStateInfo, aggregatedTask
 }
 
 func connectorsResponseToClusterConnectorInfo(configHook KafkaConnectToConsoleHook, c *con.ListConnectorsResponseExpanded) *ClusterConnectorInfo {
-	// There seems to be a type missmatch for now I think it's ok to do this conversion looking for a fix in the upstream implementation
-	tasks := make([]con.TaskState, len(c.Status.Tasks))
+	// Get the aggregated tasks status, counts and formatted errors
+	aggregatedTasksStatus := getAggregatedTasksStatus(c.Info.Name, c.Status.Tasks)
 
-	for i, task := range c.Status.Tasks {
-		tasks[i] = con.TaskState{
-			ID:       task.ID,
-			State:    task.State,
-			WorkerID: task.WorkerID,
-			Trace:    task.Trace,
-		}
-	}
-
-	status := con.ConnectorStateInfo{
-		Name:      c.Status.Name,
-		Connector: c.Status.Connector,
-		Tasks:     tasks,
-		Type:      c.Status.Type,
-	}
-
-	aggregatedTasksStatus := getAggregatedTasksStatus(c.Info.Name, tasks)
-
-	holisticState := getHolisticStateFromConnector(status, aggregatedTasksStatus)
+	// Get the holistic state, applying our custom logic
+	holisticState := getHolisticStateFromConnector(c.Status, aggregatedTasksStatus)
 
 	connectorErrors := holisticState.Errors