feat: Implemented additional REST API methods

connectors.go

@@ -148,6 +148,16 @@ func (c *Client) UpdateConnectorConfig(name string, config ConnectorConfig) (*Co
 	return connector, response, err
 }
 
+// GetConnectorTaskStatus gets the status of task for a connector.
+//
+// See: https://docs.confluent.io/current/connect/references/restapi.html#get--connectors-(string-name)-tasks-(int-taskid)-status
+func (c *Client) GetConnectorTaskStatus(name string, taskID int) (*TaskState, *http.Response, error) {
+	path := fmt.Sprintf("connectors/%v/tasks/%v/status", name, taskID)
+	status := new(TaskState)
+	respnse, err := c.get(path, status)
+	return status, respnse, err
+}
+
 // DeleteConnector deletes a connector with the given name, halting all tasks
 // and deleting its configuration.
 //
@@ -182,3 +192,11 @@ func (c *Client) RestartConnector(name string) (*http.Response, error) {
 	path := fmt.Sprintf("connectors/%v/restart", name)
 	return c.doRequest("POST", path, nil, nil)
 }
+
+// RestartConnectorTask restarts a tasks for a connector.
+//
+// See https://docs.confluent.io/current/connect/references/restapi.html#post--connectors-(string-name)-tasks-(int-taskid)-restart
+func (c *Client) RestartConnectorTask(name string, taskID int) (*http.Response, error) {
+	path := fmt.Sprintf("connectors/%v/tasks/%v/restart", name, taskID)
+	return c.doRequest("POST", path, nil, nil)
+}

connectors_test.go

@@ -422,6 +422,50 @@ var _ = Describe("Connector CRUD", func() {
 			})
 		})
 	})
+
+	Describe("GetConnectorTaskStatus", func() {
+		var statusCode int
+		resultStatus := &TaskState{
+			ID:       0,
+			State:    "RUNNING",
+			WorkerID: "127.0.0.1:8083",
+		}
+
+		BeforeEach(func() {
+			server.AppendHandlers(
+				ghttp.CombineHandlers(
+					ghttp.VerifyRequest("GET", "/connectors/local-file-source/tasks/0/status"),
+					ghttp.VerifyHeader(jsonAcceptHeader),
+					ghttp.RespondWithJSONEncodedPtr(&statusCode, &resultStatus),
+				),
+			)
+		})
+
+		Context("when existing task id is given", func() {
+			BeforeEach(func() {
+				statusCode = http.StatusOK
+			})
+
+			It("returns connector task status", func() {
+				status, _, err := client.GetConnectorTaskStatus("local-file-source", 0)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(status).To(Equal(resultStatus))
+			})
+		})
+
+		Context("when nonexisting task id is given", func() {
+			BeforeEach(func() {
+				statusCode = http.StatusNotFound
+			})
+
+			It("returns an error response", func() {
+				status, resp, err := client.GetConnectorTaskStatus("local-file-source", 0)
+				Expect(err).To(HaveOccurred())
+				Expect(*status).To(BeZero())
+				Expect(resp.StatusCode).To(Equal(http.StatusNotFound))
+			})
+		})
+	})
 })
 
 var _ = Describe("Connector Lifecycle", func() {
@@ -560,4 +604,55 @@ var _ = Describe("Connector Lifecycle", func() {
 			})
 		})
 	})
+
+	Describe("RestartConnectorTask", func() {
+		var statusCode int
+
+		BeforeEach(func() {
+			server.AppendHandlers(
+				ghttp.CombineHandlers(
+					ghttp.VerifyRequest("POST", "/connectors/local-file-source/tasks/0/restart"),
+					ghttp.VerifyHeader(jsonAcceptHeader),
+					ghttp.RespondWithPtr(&statusCode, nil),
+				),
+			)
+		})
+
+		Context("when existing task id is given", func() {
+			BeforeEach(func() {
+				statusCode = http.StatusOK
+			})
+
+			It("restarts connector", func() {
+				resp, err := client.RestartConnectorTask("local-file-source", 0)
+				Expect(err).NotTo(HaveOccurred())
+				Expect(resp.StatusCode).To(Equal(http.StatusOK))
+			})
+
+			Context("when rebalance is in process", func() {
+				BeforeEach(func() {
+					statusCode = http.StatusConflict
+				})
+
+				It("returns error with a conflict response", func() {
+					resp, err := client.RestartConnectorTask("local-file-source", 0)
+					Expect(err).To(HaveOccurred())
+					Expect(resp.StatusCode).To(Equal(http.StatusConflict))
+				})
+			})
+		})
+
+		Context("when nonexisting task id is given", func() {
+			BeforeEach(func() {
+				// The API actually throws a 500 on POST to nonexistent
+				statusCode = http.StatusInternalServerError
+			})
+
+			It("returns error with a server error response", func() {
+				resp, err := client.RestartConnectorTask("local-file-source", 0)
+				Expect(err).To(HaveOccurred())
+				Expect(resp.StatusCode).To(Equal(http.StatusInternalServerError))
+			})
+		})
+	})
 })

plugins.go

@@ -0,0 +1,21 @@
+package connect
+
+import "net/http"
+
+// Plugin represents a Kafka Connect connector plugin
+type Plugin struct {
+	Class   string `json:"class"`
+	Type    string `json:"type"`
+	Version string `json:"version"`
+}
+
+// ListPlugins retrieves a list of the installed plugins.
+//
+// See: https://docs.confluent.io/current/connect/references/restapi.html#get--connector-plugins-
+func (c *Client) ListPlugins() ([]*Plugin, *http.Response, error) {
+	path := "connector-plugins"
+	var names []*Plugin
+
+	response, err := c.get(path, &names)
+	return names, response, err
+}

plugins_test.go

@@ -0,0 +1,51 @@
+package connect_test
+
+import (
+	"net/http"
+
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+	"github.com/onsi/gomega/ghttp"
+
+	. "github.com/go-kafka/connect"
+)
+
+var _ = Describe("Plugins Tests", func() {
+	BeforeEach(func() {
+		server = ghttp.NewServer()
+		client = NewClient(server.URL())
+	})
+
+	AfterEach(func() {
+		server.Close()
+	})
+
+	Describe("ListPlugins", func() {
+		var statusCode int
+		resultPlugins := []*Plugin{
+			&Plugin{
+				Class:   "test-class",
+				Type:    "source",
+				Version: "5.3.0",
+			},
+		}
+
+		BeforeEach(func() {
+			statusCode = http.StatusOK
+
+			server.AppendHandlers(
+				ghttp.CombineHandlers(
+					ghttp.VerifyRequest("GET", "/connector-plugins"),
+					ghttp.VerifyHeader(jsonAcceptHeader),
+					ghttp.RespondWithJSONEncodedPtr(&statusCode, &resultPlugins),
+				),
+			)
+		})
+
+		It("returns list of connector plugins", func() {
+			plugins, _, err := client.ListPlugins()
+			Expect(err).NotTo(HaveOccurred())
+			Expect(plugins).To(Equal(resultPlugins))
+		})
+	})
+})