README.md

@@ -225,7 +225,6 @@ r := kafka.NewReader(kafka.ReaderConfig{
     Brokers:   []string{"localhost:9092","localhost:9093", "localhost:9094"},
     Topic:     "topic-A",
     Partition: 0,
-    MinBytes:  10e3, // 10KB
     MaxBytes:  10e6, // 10MB
 })
 r.SetOffset(42)
@@ -256,7 +255,6 @@ r := kafka.NewReader(kafka.ReaderConfig{
     Brokers:   []string{"localhost:9092", "localhost:9093", "localhost:9094"},
     GroupID:   "consumer-group-id",
     Topic:     "topic-A",
-    MinBytes:  10e3, // 10KB
     MaxBytes:  10e6, // 10MB
 })
 
@@ -320,7 +318,6 @@ r := kafka.NewReader(kafka.ReaderConfig{
     Brokers:        []string{"localhost:9092", "localhost:9093", "localhost:9094"},
     GroupID:        "consumer-group-id",
     Topic:          "topic-A",
-    MinBytes:       10e3, // 10KB
     MaxBytes:       10e6, // 10MB
     CommitInterval: time.Second, // flushes commits to Kafka every second
 })
@@ -412,6 +409,7 @@ for i := 0; i < retries; i++ {
     if err != nil {
         log.Fatalf("unexpected error %v", err)
     }
+    break
 }
 
 if err := w.Close(); err != nil {
@@ -718,7 +716,6 @@ r := kafka.NewReader(kafka.ReaderConfig{
     Brokers:   []string{"localhost:9092", "localhost:9093", "localhost:9094"},
     Topic:     "my-topic1",
     Partition: 0,
-    MinBytes:  batchSize,
     MaxBytes:  batchSize,
 })
 

alterclientquotas.go

@@ -0,0 +1,131 @@
+package kafka
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"time"
+
+	"github.com/segmentio/kafka-go/protocol/alterclientquotas"
+)
+
+// AlterClientQuotasRequest represents a request sent to a kafka broker to
+// alter client quotas.
+type AlterClientQuotasRequest struct {
+	// Address of the kafka broker to send the request to.
+	Addr net.Addr
+
+	// List of client quotas entries to alter.
+	Entries []AlterClientQuotaEntry
+
+	// Whether the alteration should be validated, but not performed.
+	ValidateOnly bool
+}
+
+type AlterClientQuotaEntry struct {
+	// The quota entities to alter.
+	Entities []AlterClientQuotaEntity
+
+	// An individual quota configuration entry to alter.
+	Ops []AlterClientQuotaOps
+}
+
+type AlterClientQuotaEntity struct {
+	// The quota entity type.
+	EntityType string
+
+	// The name of the quota entity, or null if the default.
+	EntityName string
+}
+
+type AlterClientQuotaOps struct {
+	// The quota configuration key.
+	Key string
+
+	// The quota configuration value to set, otherwise ignored if the value is to be removed.
+	Value float64
+
+	// Whether the quota configuration value should be removed, otherwise set.
+	Remove bool
+}
+
+type AlterClientQuotaResponseQuotas struct {
+	// Error is set to a non-nil value including the code and message if a top-level
+	// error was encountered when doing the update.
+	Error error
+
+	// The altered quota entities.
+	Entities []AlterClientQuotaEntity
+}
+
+// AlterClientQuotasResponse represents a response from a kafka broker to an alter client
+// quotas request.
+type AlterClientQuotasResponse struct {
+	// The amount of time that the broker throttled the request.
+	Throttle time.Duration
+
+	// List of altered client quotas responses.
+	Entries []AlterClientQuotaResponseQuotas
+}
+
+// AlterClientQuotas sends client quotas alteration request to a kafka broker and returns
+// the response.
+func (c *Client) AlterClientQuotas(ctx context.Context, req *AlterClientQuotasRequest) (*AlterClientQuotasResponse, error) {
+	entries := make([]alterclientquotas.Entry, len(req.Entries))
+
+	for entryIdx, entry := range req.Entries {
+		entities := make([]alterclientquotas.Entity, len(entry.Entities))
+		for entityIdx, entity := range entry.Entities {
+			entities[entityIdx] = alterclientquotas.Entity{
+				EntityType: entity.EntityType,
+				EntityName: entity.EntityName,
+			}
+		}
+
+		ops := make([]alterclientquotas.Ops, len(entry.Ops))
+		for opsIdx, op := range entry.Ops {
+			ops[opsIdx] = alterclientquotas.Ops{
+				Key:    op.Key,
+				Value:  op.Value,
+				Remove: op.Remove,
+			}
+		}
+
+		entries[entryIdx] = alterclientquotas.Entry{
+			Entities: entities,
+			Ops:      ops,
+		}
+	}
+
+	m, err := c.roundTrip(ctx, req.Addr, &alterclientquotas.Request{
+		Entries:      entries,
+		ValidateOnly: req.ValidateOnly,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("kafka.(*Client).AlterClientQuotas: %w", err)
+	}
+
+	res := m.(*alterclientquotas.Response)
+	responseEntries := make([]AlterClientQuotaResponseQuotas, len(res.Results))
+
+	for responseEntryIdx, responseEntry := range res.Results {
+		responseEntities := make([]AlterClientQuotaEntity, len(responseEntry.Entities))
+		for responseEntityIdx, responseEntity := range responseEntry.Entities {
+			responseEntities[responseEntityIdx] = AlterClientQuotaEntity{
+				EntityType: responseEntity.EntityType,
+				EntityName: responseEntity.EntityName,
+			}
+		}
+
+		responseEntries[responseEntryIdx] = AlterClientQuotaResponseQuotas{
+			Error:    makeError(responseEntry.ErrorCode, responseEntry.ErrorMessage),
+			Entities: responseEntities,
+		}
+	}
+	ret := &AlterClientQuotasResponse{
+		Throttle: makeDuration(res.ThrottleTimeMs),
+		Entries:  responseEntries,
+	}
+
+	return ret, nil
+}

alterclientquotas_test.go

@@ -0,0 +1,104 @@
+package kafka
+
+import (
+	"context"
+	"testing"
+
+	ktesting "github.com/segmentio/kafka-go/testing"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestClientAlterClientQuotas(t *testing.T) {
+	// Added in Version 2.6.0 https://issues.apache.org/jira/browse/KAFKA-7740
+	if !ktesting.KafkaIsAtLeast("2.6.0") {
+		return
+	}
+
+	const (
+		entityType = "client-id"
+		entityName = "my-client-id"
+		key        = "producer_byte_rate"
+		value      = 500000.0
+	)
+
+	client, shutdown := newLocalClient()
+	defer shutdown()
+
+	alterResp, err := client.AlterClientQuotas(context.Background(), &AlterClientQuotasRequest{
+		Entries: []AlterClientQuotaEntry{
+			{
+				Entities: []AlterClientQuotaEntity{
+					{
+						EntityType: entityType,
+						EntityName: entityName,
+					},
+				},
+				Ops: []AlterClientQuotaOps{
+					{
+						Key:    key,
+						Value:  value,
+						Remove: false,
+					},
+				},
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expectedAlterResp := AlterClientQuotasResponse{
+		Throttle: 0,
+		Entries: []AlterClientQuotaResponseQuotas{
+			{
+				Error: makeError(0, ""),
+				Entities: []AlterClientQuotaEntity{
+					{
+						EntityName: entityName,
+						EntityType: entityType,
+					},
+				},
+			},
+		},
+	}
+
+	assert.Equal(t, expectedAlterResp, *alterResp)
+
+	describeResp, err := client.DescribeClientQuotas(context.Background(), &DescribeClientQuotasRequest{
+		Components: []DescribeClientQuotasRequestComponent{
+			{
+				EntityType: entityType,
+				MatchType:  0,
+				Match:      entityName,
+			},
+		},
+	})
+
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	expectedDescribeResp := DescribeClientQuotasResponse{
+		Throttle: 0,
+		Error:    makeError(0, ""),
+		Entries: []DescribeClientQuotasResponseQuotas{
+			{
+				Entities: []DescribeClientQuotasEntity{
+					{
+						EntityType: entityType,
+						EntityName: entityName,
+					},
+				},
+				Values: []DescribeClientQuotasValue{
+					{
+						Key:   key,
+						Value: value,
+					},
+				},
+			},
+		},
+	}
+
+	assert.Equal(t, expectedDescribeResp, *describeResp)
+}

createtopics.go

@@ -3,7 +3,6 @@ package kafka
 import (
 	"bufio"
 	"context"
-	"errors"
 	"fmt"
 	"net"
 	"time"
@@ -65,7 +64,6 @@ func (c *Client) CreateTopics(ctx context.Context, req *CreateTopicsRequest) (*C
 		TimeoutMs:    c.timeoutMs(ctx, defaultCreateTopicsTimeout),
 		ValidateOnly: req.ValidateOnly,
 	})
-
 	if err != nil {
 		return nil, fmt.Errorf("kafka.(*Client).CreateTopics: %w", err)
 	}
@@ -363,6 +361,9 @@ func (c *Conn) createTopics(request createTopicsRequestV0) (createTopicsResponse
 		return response, err
 	}
 	for _, tr := range response.TopicErrors {
+		if tr.ErrorCode == int16(TopicAlreadyExists) {
+			continue
+		}
 		if tr.ErrorCode != 0 {
 			return response, Error(tr.ErrorCode)
 		}
@@ -385,14 +386,5 @@ func (c *Conn) CreateTopics(topics ...TopicConfig) error {
 	_, err := c.createTopics(createTopicsRequestV0{
 		Topics: requestV0Topics,
 	})
-	if err != nil {
-		if errors.Is(err, TopicAlreadyExists) {
-			// ok
-			return nil
-		}
-
-		return err
-	}
-
-	return nil
+	return err
 }