Add DescribeConfigs, AlterConfigs & CreatePartition implementation

Author: luffyao

alterconfigs.go

@@ -0,0 +1,106 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/segmentio/kafka-go/protocol/alterconfigs"
+)
+
+// AlterConfigsRequest represents a request sent to a kafka broker to alter configs
+type AlterConfigsRequest struct {
+	// Address of the kafka broker to send the request to.
+	Addr net.Addr
+
+	// List of resources to update.
+	Resources []AlterConfigRequestResource
+
+	// When set to true, topics are not created but the configuration is
+	// validated as if they were.
+	ValidateOnly bool
+}
+
+type AlterConfigRequestResource struct {
+	// Resource Type
+	ResourceType ResourceType
+
+	// Resource Name
+	ResourceName string
+
+	// Configs is a list of configuration updates.
+	Configs []AlterConfigRequestConfig
+}
+
+type AlterConfigRequestConfig struct {
+	// Configuration key name
+	Name string
+
+	// The value to set for the configuration key.
+	Value string
+}
+
+// AlterConfigsResponse represents a response from a kafka broker to an alter config request.
+type AlterConfigsResponse struct {
+	// Duration for which the request was throttled due to a quota violation.
+	Throttle time.Duration
+
+	// Mapping of topic names to errors that occurred while attempting to create
+	// the topics.
+	//
+	// The errors contain the kafka error code. Programs may use the standard
+	// errors.Is function to test the error against kafka error codes.
+	Errors map[AlterConfigsResponseResource]error
+}
+
+// AlterConfigsResponseResource
+type AlterConfigsResponseResource struct {
+	Type int8
+	Name string
+}
+
+// AlterConfigs sends a config altering request to a kafka broker and returns the
+// response.
+func (c *Client) AlterConfigs(ctx context.Context, req *AlterConfigsRequest) (*AlterConfigsResponse, error) {
+	resources := make([]alterconfigs.RequestResources, len(req.Resources))
+
+	for i, t := range req.Resources {
+		configs := make([]alterconfigs.RequestConfig, len(t.Configs))
+		for j, v := range t.Configs {
+			configs[j] = alterconfigs.RequestConfig{
+				Name:  v.Name,
+				Value: v.Value,
+			}
+		}
+		resources[i] = alterconfigs.RequestResources{
+			ResourceType: int8(t.ResourceType),
+			ResourceName: t.ResourceName,
+			Configs:      configs,
+		}
+	}
+
+	m, err := c.roundTrip(ctx, req.Addr, &alterconfigs.Request{
+		Resources:    resources,
+		ValidateOnly: req.ValidateOnly,
+	})
+
+	if err != nil {
+		return nil, fmt.Errorf("kafka.(*Client).AlterConfigs: %w", err)
+	}
+
+	res := m.(*alterconfigs.Response)
+	ret := &AlterConfigsResponse{
+		Throttle: makeDuration(res.ThrottleTimeMs),
+		Errors:   make(map[AlterConfigsResponseResource]error, len(res.Responses)),
+	}
+
+	for _, t := range res.Responses {
+		ret.Errors[AlterConfigsResponseResource{
+			Type: t.ResourceType,
+			Name: t.ResourceName,
+		}] = makeError(t.ErrorCode, t.ErrorMessage)
+	}
+
+	return ret, nil
+}

alterconfigs_test.go

@@ -0,0 +1,63 @@
+package kafka
+
+import (
+	"context"
+	"testing"
+
+	ktesting "github.com/segmentio/kafka-go/testing"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClientAlterConfigs(t *testing.T) {
+	if !ktesting.KafkaIsAtLeast("0.11.0") {
+		return
+	}
+
+	const (
+		MaxMessageBytes      = "max.message.bytes"
+		MaxMessageBytesValue = "200000"
+	)
+
+	client, shutdown := newLocalClient()
+	defer shutdown()
+
+	topic := makeTopic()
+	createTopic(t, topic, 1)
+	defer deleteTopic(t, topic)
+
+	_, err := client.AlterConfigs(context.Background(), &AlterConfigsRequest{
+		Resources: []AlterConfigRequestResource{{
+			ResourceType: ResourceTypeTopic,
+			ResourceName: topic,
+			Configs: []AlterConfigRequestConfig{{
+				Name:  MaxMessageBytes,
+				Value: MaxMessageBytesValue,
+			},
+			},
+		}},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	describeResp, err := client.DescribeConfigs(context.Background(), &DescribeConfigsRequest{
+		Resources: []DescribeConfigRequestResource{{
+			ResourceType: ResourceTypeTopic,
+			ResourceName: topic,
+			ConfigNames:  []string{MaxMessageBytes},
+		}},
+	})
+
+	maxMessageBytesValue := "0"
+	for _, resource := range describeResp.Resources {
+		if resource.ResourceType == int8(ResourceTypeTopic) && resource.ResourceName == topic {
+			for _, entry := range resource.ConfigEntries {
+				if entry.ConfigName == MaxMessageBytes {
+					maxMessageBytesValue = entry.ConfigValue
+				}
+			}
+		}
+	}
+	assert.Equal(t, maxMessageBytesValue, MaxMessageBytesValue)
+}

client.go

@@ -10,10 +10,11 @@ import (
 )
 
 const (
-	defaultCreateTopicsTimeout = 2 * time.Second
-	defaultDeleteTopicsTimeout = 2 * time.Second
-	defaultProduceTimeout      = 500 * time.Millisecond
-	defaultMaxWait             = 500 * time.Millisecond
+	defaultCreateTopicsTimeout     = 2 * time.Second
+	defaultDeleteTopicsTimeout     = 2 * time.Second
+	defaultCreatePartitionsTimeout = 2 * time.Second
+	defaultProduceTimeout          = 500 * time.Millisecond
+	defaultMaxWait                 = 500 * time.Millisecond
 )
 
 // Client is a high-level API to interract with kafka brokers.

createpartitions.go

@@ -0,0 +1,103 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/segmentio/kafka-go/protocol/createpartitions"
+)
+
+// CreatePartitionsRequest represents a request sent to a kafka broker to create
+// and update topic parititions.
+type CreatePartitionsRequest struct {
+	// Address of the kafka broker to send the request to.
+	Addr net.Addr
+
+	// List of topics to create and their configuration.
+	Topics []TopicPartitionsConfig
+
+	// When set to true, topics are not created but the configuration is
+	// validated as if they were.
+	ValidateOnly bool
+}
+
+// CreatePartitionsResponse represents a response from a kafka broker to a partition
+// creation request.
+type CreatePartitionsResponse struct {
+	// The amount of time that the broker throttled the request.
+	Throttle time.Duration
+
+	// Mapping of topic names to errors that occurred while attempting to create
+	// the topics.
+	//
+	// The errors contain the kafka error code. Programs may use the standard
+	// errors.Is function to test the error against kafka error codes.
+	Errors map[string]error
+}
+
+// CreatePartitions sends a partitions creation request to a kafka broker and returns the
+// response.
+func (c *Client) CreatePartitions(ctx context.Context, req *CreatePartitionsRequest) (*CreatePartitionsResponse, error) {
+	topics := make([]createpartitions.RequestTopic, len(req.Topics))
+
+	for i, t := range req.Topics {
+		topics[i] = createpartitions.RequestTopic{
+			Name:        t.Name,
+			Count:       t.Count,
+			Assignments: t.assignments(),
+		}
+	}
+
+	m, err := c.roundTrip(ctx, req.Addr, &createpartitions.Request{
+		Topics:       topics,
+		TimeoutMs:    c.timeoutMs(ctx, defaultCreatePartitionsTimeout),
+		ValidateOnly: req.ValidateOnly,
+	})
+
+	if err != nil {
+		return nil, fmt.Errorf("kafka.(*Client).CreatePartitions: %w", err)
+	}
+
+	res := m.(*createpartitions.Response)
+	ret := &CreatePartitionsResponse{
+		Throttle: makeDuration(res.ThrottleTimeMs),
+		Errors:   make(map[string]error, len(res.Results)),
+	}
+
+	for _, t := range res.Results {
+		ret.Errors[t.Name] = makeError(t.ErrorCode, t.ErrorMessage)
+	}
+
+	return ret, nil
+}
+
+type TopicPartitionsConfig struct {
+	// Topic name
+	Name string
+
+	// Topic partition's count.
+	Count int32
+
+	// TopicPartitionAssignments among kafka brokers for this topic partitions.
+	TopicPartitionAssignments []TopicPartitionAssignment
+}
+
+func (t *TopicPartitionsConfig) assignments() []createpartitions.RequestAssignment {
+	if len(t.TopicPartitionAssignments) == 0 {
+		return nil
+	}
+	assignments := make([]createpartitions.RequestAssignment, len(t.TopicPartitionAssignments))
+	for i, a := range t.TopicPartitionAssignments {
+		assignments[i] = createpartitions.RequestAssignment{
+			BrokerIDs: a.BrokerIDs,
+		}
+	}
+	return assignments
+}
+
+type TopicPartitionAssignment struct {
+	// Broker IDs
+	BrokerIDs []int32
+}

createpartitions_test.go

@@ -0,0 +1,44 @@
+package kafka
+
+import (
+	"context"
+	"testing"
+
+	ktesting "github.com/segmentio/kafka-go/testing"
+)
+
+func TestClientCreatePartitions(t *testing.T) {
+	if !ktesting.KafkaIsAtLeast("1.0.1") {
+		return
+	}
+
+	client, shutdown := newLocalClient()
+	defer shutdown()
+
+	topic := makeTopic()
+	createTopic(t, topic, 1)
+	defer deleteTopic(t, topic)
+
+	res, err := client.CreatePartitions(context.Background(), &CreatePartitionsRequest{
+		Topics: []TopicPartitionsConfig{
+			TopicPartitionsConfig{
+				Name:  topic,
+				Count: 2,
+				TopicPartitionAssignments: []TopicPartitionAssignment{
+					TopicPartitionAssignment{
+						BrokerIDs: []int32{1},
+					},
+				},
+			},
+		},
+		ValidateOnly: false,
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	if err := res.Errors[topic]; err != nil {
+		t.Error(err)
+	}
+}