Repository: Shopify/sarama
Pull request #1966 (2021-06-14 08:57:40)

KIP-339: Add Incremental Config updates API
Hello!

In this PR I cover IncrementalAlterConfigs API in Sarama. It is very useful, due to fact we can specify which config entry needs to be affected (instead of sending everything over and over again).

Fixes: #1830, https://github.com/birdayz/kaf/issues/155

Kafka Code Links:
- [KIP-339](https://cwiki.apache.org/confluence/display/KAFKA/KIP-339%3A+Create+a+new+IncrementalAlterConfigs+API)
- Operation definitions: https://github.com/apache/kafka/blob/2.3/clients/src/main/java/org/apache/kafka/clients/admin/AlterConfigOp.java#L38

Best Regards,
Adam.


Patch:
diff --git a/alter_configs_response.go b/alter_configs_response.go
index 89065ab..cfb6369 100644
--- a/alter_configs_response.go
+++ b/alter_configs_response.go
@@ -23,16 +23,9 @@ func (a *AlterConfigsResponse) encode(pe packetEncoder) error {
 		return err
 	}
 
-	for i := range a.Resources {
-		pe.putInt16(a.Resources[i].ErrorCode)
-		err := pe.putString(a.Resources[i].ErrorMsg)
-		if err != nil {
-			return nil
-		}
-		pe.putInt8(int8(a.Resources[i].Type))
-		err = pe.putString(a.Resources[i].Name)
-		if err != nil {
-			return nil
+	for _, v := range a.Resources {
+		if err := v.encode(pe); err != nil {
+			return err
 		}
 	}
 
@@ -56,30 +49,52 @@ func (a *AlterConfigsResponse) decode(pd packetDecoder, version int16) error {
 	for i := range a.Resources {
 		a.Resources[i] = new(AlterConfigsResourceResponse)
 
-		errCode, err := pd.getInt16()
-		if err != nil {
+		if err := a.Resources[i].decode(pd, version); err != nil {
 			return err
 		}
-		a.Resources[i].ErrorCode = errCode
+	}
 
-		e, err := pd.getString()
-		if err != nil {
-			return err
-		}
-		a.Resources[i].ErrorMsg = e
+	return nil
+}
 
-		t, err := pd.getInt8()
-		if err != nil {
-			return err
-		}
-		a.Resources[i].Type = ConfigResourceType(t)
+func (a *AlterConfigsResourceResponse) encode(pe packetEncoder) error {
+	pe.putInt16(a.ErrorCode)
+	err := pe.putString(a.ErrorMsg)
+	if err != nil {
+		return nil
+	}
+	pe.putInt8(int8(a.Type))
+	err = pe.putString(a.Name)
+	if err != nil {
+		return nil
+	}
+	return nil
+}
 
-		name, err := pd.getString()
-		if err != nil {
-			return err
-		}
-		a.Resources[i].Name = name
+func (a *AlterConfigsResourceResponse) decode(pd packetDecoder, version int16) error {
+	errCode, err := pd.getInt16()
+	if err != nil {
+		return err
+	}
+	a.ErrorCode = errCode
+
+	e, err := pd.getString()
+	if err != nil {
+		return err
+	}
+	a.ErrorMsg = e
+
+	t, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.Type = ConfigResourceType(t)
+
+	name, err := pd.getString()
+	if err != nil {
+		return err
 	}
+	a.Name = name
 
 	return nil
 }
diff --git a/broker.go b/broker.go
index 664b74f..dd01e4e 100644
--- a/broker.go
+++ b/broker.go
@@ -666,6 +666,18 @@ func (b *Broker) AlterConfigs(request *AlterConfigsRequest) (*AlterConfigsRespon
 	return response, nil
 }
 
+// IncrementalAlterConfigs sends a request to incremental alter config and return a response or error
+func (b *Broker) IncrementalAlterConfigs(request *IncrementalAlterConfigsRequest) (*IncrementalAlterConfigsResponse, error) {
+	response := new(IncrementalAlterConfigsResponse)
+
+	err := b.sendAndReceive(request, response)
+	if err != nil {
+		return nil, err
+	}
+
+	return response, nil
+}
+
 // DeleteGroups sends a request to delete groups and returns a response or error
 func (b *Broker) DeleteGroups(request *DeleteGroupsRequest) (*DeleteGroupsResponse, error) {
 	response := new(DeleteGroupsResponse)
diff --git a/incremental_alter_configs_request.go b/incremental_alter_configs_request.go
new file mode 100644
index 0000000..c4d05a9
--- /dev/null
+++ b/incremental_alter_configs_request.go
@@ -0,0 +1,173 @@
+package sarama
+
+type IncrementalAlterConfigsOperation int8
+
+const (
+	IncrementalAlterConfigsOperationSet IncrementalAlterConfigsOperation = iota
+	IncrementalAlterConfigsOperationDelete
+	IncrementalAlterConfigsOperationAppend
+	IncrementalAlterConfigsOperationSubtract
+)
+
+// IncrementalAlterConfigsRequest is an incremental alter config request type
+type IncrementalAlterConfigsRequest struct {
+	Resources    []*IncrementalAlterConfigsResource
+	ValidateOnly bool
+}
+
+type IncrementalAlterConfigsResource struct {
+	Type          ConfigResourceType
+	Name          string
+	ConfigEntries map[string]IncrementalAlterConfigsEntry
+}
+
+type IncrementalAlterConfigsEntry struct {
+	Operation IncrementalAlterConfigsOperation
+	Value     *string
+}
+
+func (a *IncrementalAlterConfigsRequest) encode(pe packetEncoder) error {
+	if err := pe.putArrayLength(len(a.Resources)); err != nil {
+		return err
+	}
+
+	for _, r := range a.Resources {
+		if err := r.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	pe.putBool(a.ValidateOnly)
+	return nil
+}
+
+func (a *IncrementalAlterConfigsRequest) decode(pd packetDecoder, version int16) error {
+	resourceCount, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	a.Resources = make([]*IncrementalAlterConfigsResource, resourceCount)
+	for i := range a.Resources {
+		r := &IncrementalAlterConfigsResource{}
+		err = r.decode(pd, version)
+		if err != nil {
+			return err
+		}
+		a.Resources[i] = r
+	}
+
+	validateOnly, err := pd.getBool()
+	if err != nil {
+		return err
+	}
+
+	a.ValidateOnly = validateOnly
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsResource) encode(pe packetEncoder) error {
+	pe.putInt8(int8(a.Type))
+
+	if err := pe.putString(a.Name); err != nil {
+		return err
+	}
+
+	if err := pe.putArrayLength(len(a.ConfigEntries)); err != nil {
+		return err
+	}
+
+	for name, e := range a.ConfigEntries {
+		if err := pe.putString(name); err != nil {
+			return err
+		}
+
+		if err := e.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsResource) decode(pd packetDecoder, version int16) error {
+	t, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.Type = ConfigResourceType(t)
+
+	name, err := pd.getString()
+	if err != nil {
+		return err
+	}
+	a.Name = name
+
+	n, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	if n > 0 {
+		a.ConfigEntries = make(map[string]IncrementalAlterConfigsEntry, n)
+		for i := 0; i < n; i++ {
+			name, err := pd.getString()
+			if err != nil {
+				return err
+			}
+
+			var v IncrementalAlterConfigsEntry
+
+			if err := v.decode(pd, version); err != nil {
+				return err
+			}
+
+			a.ConfigEntries[name] = v
+		}
+	}
+	return err
+}
+
+func (a *IncrementalAlterConfigsEntry) encode(pe packetEncoder) error {
+	pe.putInt8(int8(a.Operation))
+
+	if err := pe.putNullableString(a.Value); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsEntry) decode(pd packetDecoder, version int16) error {
+	t, err := pd.getInt8()
+	if err != nil {
+		return err
+	}
+	a.Operation = IncrementalAlterConfigsOperation(t)
+
+	s, err := pd.getNullableString()
+	if err != nil {
+		return err
+	}
+
+	a.Value = s
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsRequest) key() int16 {
+	return 44
+}
+
+func (a *IncrementalAlterConfigsRequest) version() int16 {
+	return 0
+}
+
+func (a *IncrementalAlterConfigsRequest) headerVersion() int16 {
+	return 1
+}
+
+func (a *IncrementalAlterConfigsRequest) requiredVersion() KafkaVersion {
+	return V2_3_0_0
+}
diff --git a/incremental_alter_configs_response.go b/incremental_alter_configs_response.go
new file mode 100644
index 0000000..3e8c450
--- /dev/null
+++ b/incremental_alter_configs_response.go
@@ -0,0 +1,66 @@
+package sarama
+
+import "time"
+
+// IncrementalAlterConfigsResponse is a response type for incremental alter config
+type IncrementalAlterConfigsResponse struct {
+	ThrottleTime time.Duration
+	Resources    []*AlterConfigsResourceResponse
+}
+
+func (a *IncrementalAlterConfigsResponse) encode(pe packetEncoder) error {
+	pe.putInt32(int32(a.ThrottleTime / time.Millisecond))
+
+	if err := pe.putArrayLength(len(a.Resources)); err != nil {
+		return err
+	}
+
+	for _, v := range a.Resources {
+		if err := v.encode(pe); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsResponse) decode(pd packetDecoder, version int16) error {
+	throttleTime, err := pd.getInt32()
+	if err != nil {
+		return err
+	}
+	a.ThrottleTime = time.Duration(throttleTime) * time.Millisecond
+
+	responseCount, err := pd.getArrayLength()
+	if err != nil {
+		return err
+	}
+
+	a.Resources = make([]*AlterConfigsResourceResponse, responseCount)
+
+	for i := range a.Resources {
+		a.Resources[i] = new(AlterConfigsResourceResponse)
+
+		if err := a.Resources[i].decode(pd, version); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (a *IncrementalAlterConfigsResponse) key() int16 {
+	return 44
+}
+
+func (a *IncrementalAlterConfigsResponse) version() int16 {
+	return 0
+}
+
+func (a *IncrementalAlterConfigsResponse) headerVersion() int16 {
+	return 0
+}
+
+func (a *IncrementalAlterConfigsResponse) requiredVersion() KafkaVersion {
+	return V2_3_0_0
+}
diff --git a/request.go b/request.go
index afa64ff..d899df5 100644
--- a/request.go
+++ b/request.go
@@ -182,6 +182,8 @@ func allocateBody(key, version int16) protocolBody {
 		return &CreatePartitionsRequest{}
 	case 42:
 		return &DeleteGroupsRequest{}
+	case 44:
+		return &IncrementalAlterConfigsRequest{}
 	case 45:
 		return &AlterPartitionReassignmentsRequest{}
 	case 46:


Test patch:
diff --git a/incremental_alter_configs_request_test.go b/incremental_alter_configs_request_test.go
new file mode 100644
index 0000000..d239bf5
--- /dev/null
+++ b/incremental_alter_configs_request_test.go
@@ -0,0 +1,98 @@
+package sarama
+
+import "testing"
+
+var (
+	emptyIncrementalAlterConfigsRequest = []byte{
+		0, 0, 0, 0, // 0 configs
+		0, // don't Validate
+	}
+
+	singleIncrementalAlterConfigsRequest = []byte{
+		0, 0, 0, 1, // 1 config
+		2,                   // a topic
+		0, 3, 'f', 'o', 'o', // topic name: foo
+		0, 0, 0, 1, // 1 config name
+		0, 10, // 10 chars
+		's', 'e', 'g', 'm', 'e', 'n', 't', '.', 'm', 's',
+		0, // OperationSet
+		0, 4,
+		'1', '0', '0', '0',
+		0, // don't validate
+	}
+
+	doubleIncrementalAlterConfigsRequest = []byte{
+		0, 0, 0, 2, // 2 config
+		2,                   // a topic
+		0, 3, 'f', 'o', 'o', // topic name: foo
+		0, 0, 0, 1, // 1 config name
+		0, 10, // 10 chars
+		's', 'e', 'g', 'm', 'e', 'n', 't', '.', 'm', 's',
+		0, // OperationSet
+		0, 4,
+		'1', '0', '0', '0',
+		2,                   // a topic
+		0, 3, 'b', 'a', 'r', // topic name: foo
+		0, 0, 0, 1, // 2 config
+		0, 12, // 12 chars
+		'r', 'e', 't', 'e', 'n', 't', 'i', 'o', 'n', '.', 'm', 's',
+		1, // OperationDelete
+		0, 4,
+		'1', '0', '0', '0',
+		0, // don't validate
+	}
+)
+
+func TestIncrementalAlterConfigsRequest(t *testing.T) {
+	var request *IncrementalAlterConfigsRequest
+
+	request = &IncrementalAlterConfigsRequest{
+		Resources: []*IncrementalAlterConfigsResource{},
+	}
+	testRequest(t, "no requests", request, emptyIncrementalAlterConfigsRequest)
+
+	configValue := "1000"
+	request = &IncrementalAlterConfigsRequest{
+		Resources: []*IncrementalAlterConfigsResource{
+			{
+				Type: TopicResource,
+				Name: "foo",
+				ConfigEntries: map[string]IncrementalAlterConfigsEntry{
+					"segment.ms": {
+						Operation: IncrementalAlterConfigsOperationSet,
+						Value:     &configValue,
+					},
+				},
+			},
+		},
+	}
+
+	testRequest(t, "one config", request, singleIncrementalAlterConfigsRequest)
+
+	request = &IncrementalAlterConfigsRequest{
+		Resources: []*IncrementalAlterConfigsResource{
+			{
+				Type: TopicResource,
+				Name: "foo",
+				ConfigEntries: map[string]IncrementalAlterConfigsEntry{
+					"segment.ms": {
+						Operation: IncrementalAlterConfigsOperationSet,
+						Value:     &configValue,
+					},
+				},
+			},
+			{
+				Type: TopicResource,
+				Name: "bar",
+				ConfigEntries: map[string]IncrementalAlterConfigsEntry{
+					"retention.ms": {
+						Operation: IncrementalAlterConfigsOperationDelete,
+						Value:     &configValue,
+					},
+				},
+			},
+		},
+	}
+
+	testRequest(t, "two configs", request, doubleIncrementalAlterConfigsRequest)
+}
diff --git a/incremental_alter_configs_response_test.go b/incremental_alter_configs_response_test.go
new file mode 100644
index 0000000..6971567
--- /dev/null
+++ b/incremental_alter_configs_response_test.go
@@ -0,0 +1,45 @@
+package sarama
+
+import (
+	"testing"
+)
+
+var (
+	incrementalAlterResponseEmpty = []byte{
+		0, 0, 0, 0, // throttle
+		0, 0, 0, 0, // no configs
+	}
+
+	incrementalAlterResponsePopulated = []byte{
+		0, 0, 0, 0, // throttle
+		0, 0, 0, 1, // response
+		0, 0, // errorcode
+		0, 0, // string
+		2, // topic
+		0, 3, 'f', 'o', 'o',
+	}
+)
+
+func TestIncrementalAlterConfigsResponse(t *testing.T) {
+	var response *IncrementalAlterConfigsResponse
+
+	response = &IncrementalAlterConfigsResponse{
+		Resources: []*AlterConfigsResourceResponse{},
+	}
+	testVersionDecodable(t, "empty", response, incrementalAlterResponseEmpty, 0)
+	if len(response.Resources) != 0 {
+		t.Error("Expected no groups")
+	}
+
+	response = &IncrementalAlterConfigsResponse{
+		Resources: []*AlterConfigsResourceResponse{
+			{
+				ErrorCode: 0,
+				ErrorMsg:  "",
+				Type:      TopicResource,
+				Name:      "foo",
+			},
+		},
+	}
+	testResponse(t, "response with error", response, incrementalAlterResponsePopulated)
+}