Add kafka configuration to pub/sub (dapr#3275)

Signed-off-by: yaron2 <[email protected]>

Author: yaron2

common/component/kafka/kafka.go

@@ -78,6 +78,9 @@ func (k *Kafka) Init(ctx context.Context, metadata map[string]string) error {
 	config := sarama.NewConfig()
 	config.Version = meta.internalVersion
 	config.Consumer.Offsets.Initial = k.initialOffset
+	config.Consumer.Fetch.Min = meta.consumerFetchMin
+	config.Consumer.Fetch.Default = meta.consumerFetchDefault
+	config.ChannelBufferSize = meta.channelBufferSize
 
 	if meta.ClientID != "" {
 		config.ClientID = meta.ClientID

common/component/kafka/metadata.go

@@ -41,6 +41,9 @@ const (
 	mtlsAuthType         = "mtls"
 	awsIAMAuthType       = "awsiam"
 	noAuthType           = "none"
+	consumerFetchMin     = "consumerFetchMin"
+	consumerFetchDefault = "consumerFetchDefault"
+	channelBufferSize    = "channelBufferSize"
 )
 
 type KafkaMetadata struct {
@@ -72,10 +75,13 @@ type KafkaMetadata struct {
 	internalVersion        sarama.KafkaVersion `mapstructure:"-"`
 	internalOidcExtensions map[string]string   `mapstructure:"-"`
 	// aws iam auth profile
-	AWSAccessKey    string `mapstructure:"awsAccessKey"`
-	AWSSecretKey    string `mapstructure:"awsSecretKey"`
-	AWSSessionToken string `mapstructure:"awsSessionToken"`
-	AWSRegion       string `mapstructure:"awsRegion"`
+	AWSAccessKey         string `mapstructure:"awsAccessKey"`
+	AWSSecretKey         string `mapstructure:"awsSecretKey"`
+	AWSSessionToken      string `mapstructure:"awsSessionToken"`
+	AWSRegion            string `mapstructure:"awsRegion"`
+	channelBufferSize    int    `mapstructure:"-"`
+	consumerFetchMin     int32  `mapstructure:"-"`
+	consumerFetchDefault int32  `mapstructure:"-"`
 }
 
 // upgradeMetadata updates metadata properties based on deprecated usage.
@@ -119,6 +125,9 @@ func (k *Kafka) getKafkaMetadata(meta map[string]string) (*KafkaMetadata, error)
 	m := KafkaMetadata{
 		ConsumeRetryInterval: 100 * time.Millisecond,
 		internalVersion:      sarama.V2_0_0_0, //nolint:nosnakecase
+		channelBufferSize:    256,
+		consumerFetchMin:     1,
+		consumerFetchDefault: 1024 * 1024,
 	}
 
 	err := metadata.DecodeMetadata(meta, &m)
@@ -257,5 +266,32 @@ func (k *Kafka) getKafkaMetadata(meta map[string]string) (*KafkaMetadata, error)
 		m.internalVersion = version
 	}
 
+	if val, ok := meta[channelBufferSize]; ok && val != "" {
+		v, err := strconv.Atoi(val)
+		if err != nil {
+			return nil, err
+		}
+
+		m.channelBufferSize = v
+	}
+
+	if val, ok := meta[consumerFetchDefault]; ok && val != "" {
+		v, err := strconv.ParseInt(val, 10, 32)
+		if err != nil {
+			return nil, err
+		}
+
+		m.consumerFetchDefault = int32(v)
+	}
+
+	if val, ok := meta[consumerFetchMin]; ok && val != "" {
+		v, err := strconv.ParseInt(val, 10, 32)
+		if err != nil {
+			return nil, err
+		}
+
+		m.consumerFetchMin = int32(v)
+	}
+
 	return &m, nil
 }

common/component/kafka/metadata_test.go

@@ -57,6 +57,9 @@ func getCompleteMetadata() map[string]string {
 		clientKey:              clientKeyMock,
 		caCert:                 caCertMock,
 		"consumeRetryInterval": "200",
+		consumerFetchDefault:   "1048576",
+		consumerFetchMin:       "1",
+		channelBufferSize:      "256",
 	}
 }
 
@@ -125,6 +128,9 @@ func assertMetadata(t *testing.T, meta *KafkaMetadata) {
 	require.Equal(t, clientKeyMock, meta.TLSClientKey)
 	require.Equal(t, caCertMock, meta.TLSCaCert)
 	require.Equal(t, 200*time.Millisecond, meta.ConsumeRetryInterval)
+	require.Equal(t, int32(1024*1024), meta.consumerFetchDefault)
+	require.Equal(t, int32(1), meta.consumerFetchMin)
+	require.Equal(t, 256, meta.channelBufferSize)
 }
 
 func TestMissingBrokers(t *testing.T) {
@@ -361,3 +367,25 @@ func TestTls(t *testing.T) {
 		require.Equal(t, "missing CA certificate property 'caCert' for authType 'certificate'", err.Error())
 	})
 }
+
+func TestMetadataConsumerFetchValues(t *testing.T) {
+	k := getKafka()
+	m := getCompleteMetadata()
+	m["consumerFetchMin"] = "3"
+	m["consumerFetchDefault"] = "2048"
+
+	meta, err := k.getKafkaMetadata(m)
+	require.NoError(t, err)
+	require.Equal(t, int32(3), meta.consumerFetchMin)
+	require.Equal(t, int32(2048), meta.consumerFetchDefault)
+}
+
+func TestMetadataChannelBufferSize(t *testing.T) {
+	k := getKafka()
+	m := getCompleteMetadata()
+	m["channelBufferSize"] = "128"
+
+	meta, err := k.getKafkaMetadata(m)
+	require.NoError(t, err)
+	require.Equal(t, 128, meta.channelBufferSize)
+}

pubsub/kafka/metadata.yaml

@@ -241,4 +241,22 @@ metadata:
         This is potentially insecure and not recommended for use in production.
       example: "true"
       default: "false"
-      type: bool
+      type: bool
+    - name: channelBufferSize
+      type: number
+      description: |
+        The number of events to buffer in internal and external channels.
+      example: '128'
+      default: '256'
+    - name: consumerFetchMin
+      type: number
+      description: |
+        The minimum number of message bytes to fetch in a request.
+      example: '4'
+      default: '1'
+    - name: consumerFetchDefault
+      type: number
+      description: |
+        The default number of message bytes to fetch from the broker in each request.
+      example: '2097152'
+      default: '1048576'