From 76d76958328f0232c522d9602573343f342a96ef Mon Sep 17 00:00:00 2001 From: akshitguptaa Date: Wed, 8 Jul 2026 02:01:49 +0530 Subject: [PATCH 1/3] kafka: add numPartitions support for auto topic creation Signed-off-by: akshitguptaa --- common/component/kafka/kafka.go | 29 ++-- common/component/kafka/metadata.go | 12 ++ common/component/kafka/metadata_test.go | 58 +++++++ common/component/kafka/mocks/clusteradmin.go | 151 ++++++++++++++++++ common/component/kafka/topics.go | 49 ++++++ common/component/kafka/topics_test.go | 155 +++++++++++++++++++ 6 files changed, 444 insertions(+), 10 deletions(-) create mode 100644 common/component/kafka/mocks/clusteradmin.go create mode 100644 common/component/kafka/topics.go create mode 100644 common/component/kafka/topics_test.go diff --git a/common/component/kafka/kafka.go b/common/component/kafka/kafka.go index 5372d44982..8619e90921 100644 --- a/common/component/kafka/kafka.go +++ b/common/component/kafka/kafka.go @@ -45,16 +45,19 @@ type Kafka struct { mockProducer sarama.SyncProducer clients *clients - maxMessageBytes int - consumerGroup string - brokers []string - logger logger.Logger - authType string - saslUsername string - saslPassword string - initialOffset int64 - config *sarama.Config - escapeHeaders bool + maxMessageBytes int + numPartitions int32 + replicationFactor int16 + ensuredTopics sync.Map + consumerGroup string + brokers []string + logger logger.Logger + authType string + saslUsername string + saslPassword string + initialOffset int64 + config *sarama.Config + escapeHeaders bool subscribeTopics TopicHandlerConfig subscribeLock sync.Mutex @@ -223,6 +226,9 @@ func (k *Kafka) Init(ctx context.Context, metadata map[string]string) error { sarama.Logger = SaramaLogBridge{daprLogger: k.logger} k.maxMessageBytes = meta.MaxMessageBytes + k.numPartitions = meta.NumPartitions + k.replicationFactor = meta.ReplicationFactor + // Default retry configuration is used if no // backOff properties are set. if rerr := retry.DecodeConfigWithPrefix( @@ -379,6 +385,9 @@ func (k *Kafka) Close() error { errs[1] = k.clients.consumerGroup.Close() k.clients.consumerGroup = nil } + if k.clients.admin != nil { + errs[2] = k.clients.admin.Close() + } } } diff --git a/common/component/kafka/metadata.go b/common/component/kafka/metadata.go index 18b3ffd2f5..3a13711c1a 100644 --- a/common/component/kafka/metadata.go +++ b/common/component/kafka/metadata.go @@ -78,6 +78,8 @@ type KafkaMetadata struct { InitialOffset string `mapstructure:"initialOffset"` internalInitialOffset int64 `mapstructure:"-"` MaxMessageBytes int `mapstructure:"maxMessageBytes"` + NumPartitions int32 `mapstructure:"numPartitions"` + ReplicationFactor int16 `mapstructure:"replicationFactor"` OidcTokenEndpoint string `mapstructure:"oidcTokenEndpoint"` OidcClientID string `mapstructure:"oidcClientID"` OidcClientSecret string `mapstructure:"oidcClientSecret"` @@ -377,6 +379,16 @@ func (k *Kafka) getKafkaMetadata(meta map[string]string) (*KafkaMetadata, error) m.consumerFetchMin = int32(v) } + if m.NumPartitions < 0 { + return nil, errors.New("kafka error: 'numPartitions' must be a non-negative number") + } + if m.NumPartitions > 0 && strings.ToLower(m.AuthType) == awsIAMAuthType { + return nil, errors.New("kafka error: 'numPartitions' auto-topic-creation is not supported with authType 'awsiam'") + } + if m.NumPartitions > 0 && m.ReplicationFactor <= 0 { + m.ReplicationFactor = 1 + } + // confirm client connection fields are valid if m.ClientConnectionTopicMetadataRefreshInterval <= 0 { m.ClientConnectionTopicMetadataRefreshInterval = defaultClientConnectionTopicMetadataRefreshInterval diff --git a/common/component/kafka/metadata_test.go b/common/component/kafka/metadata_test.go index a06a7fb966..2ff08ec69d 100644 --- a/common/component/kafka/metadata_test.go +++ b/common/component/kafka/metadata_test.go @@ -475,6 +475,64 @@ func TestMetadataProducerValues(t *testing.T) { }) } +func TestMetadataNumPartitionsValues(t *testing.T) { + t.Run("numPartitions and replicationFactor parsed correctly", func(t *testing.T) { + k := getKafka() + m := getCompleteMetadata() + m["numPartitions"] = "6" + m["replicationFactor"] = "3" + + meta, err := k.getKafkaMetadata(m) + require.NoError(t, err) + require.Equal(t, int32(6), meta.NumPartitions) + require.Equal(t, int16(3), meta.ReplicationFactor) + }) + + t.Run("replicationFactor defaults to 1 when numPartitions is set", func(t *testing.T) { + k := getKafka() + m := getCompleteMetadata() + m["numPartitions"] = "3" + + meta, err := k.getKafkaMetadata(m) + require.NoError(t, err) + require.Equal(t, int32(3), meta.NumPartitions) + require.Equal(t, int16(1), meta.ReplicationFactor) + }) + + t.Run("numPartitions zero means no auto-create", func(t *testing.T) { + k := getKafka() + m := getCompleteMetadata() + // numPartitions defaults to 0, replicationFactor should stay 0 + meta, err := k.getKafkaMetadata(m) + require.NoError(t, err) + require.Equal(t, int32(0), meta.NumPartitions) + require.Equal(t, int16(0), meta.ReplicationFactor) + }) + + t.Run("negative numPartitions is rejected", func(t *testing.T) { + k := getKafka() + m := getCompleteMetadata() + m["numPartitions"] = "-1" + + meta, err := k.getKafkaMetadata(m) + require.Error(t, err) + require.Nil(t, meta) + require.Equal(t, "kafka error: 'numPartitions' must be a non-negative number", err.Error()) + }) + + t.Run("numPartitions with awsiam authType is rejected", func(t *testing.T) { + k := getKafka() + m := getCompleteMetadata() + m["numPartitions"] = "3" + m["authType"] = awsIAMAuthType + + meta, err := k.getKafkaMetadata(m) + require.Error(t, err) + require.Nil(t, meta) + require.Equal(t, "kafka error: 'numPartitions' auto-topic-creation is not supported with authType 'awsiam'", err.Error()) + }) +} + func TestMetadataChannelBufferSize(t *testing.T) { k := getKafka() m := getCompleteMetadata() diff --git a/common/component/kafka/mocks/clusteradmin.go b/common/component/kafka/mocks/clusteradmin.go new file mode 100644 index 0000000000..bda5e5b6bc --- /dev/null +++ b/common/component/kafka/mocks/clusteradmin.go @@ -0,0 +1,151 @@ +/* +Copyright 2026 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package mocks + +import "github.com/IBM/sarama" + +// FakeClusterAdmin implements sarama.ClusterAdmin for unit testing. +// Only CreateTopic and Close carry injectable behaviour; every other method +// returns a sensible zero-value so the mock satisfies the full interface. +type FakeClusterAdmin struct { + createTopicFn func(topic string, detail *sarama.TopicDetail, validateOnly bool) error + closeFn func() error +} + +func NewClusterAdmin() *FakeClusterAdmin { + return &FakeClusterAdmin{ + createTopicFn: func(string, *sarama.TopicDetail, bool) error { + return nil + }, + closeFn: func() error { + return nil + }, + } +} + +func (f *FakeClusterAdmin) WithCreateTopicFn(fn func(string, *sarama.TopicDetail, bool) error) *FakeClusterAdmin { + f.createTopicFn = fn + return f +} + +func (f *FakeClusterAdmin) WithCloseFn(fn func() error) *FakeClusterAdmin { + f.closeFn = fn + return f +} + +// --- sarama.ClusterAdmin interface --- + +func (f *FakeClusterAdmin) CreateTopic(topic string, detail *sarama.TopicDetail, validateOnly bool) error { + return f.createTopicFn(topic, detail, validateOnly) +} + +func (f *FakeClusterAdmin) ListTopics() (map[string]sarama.TopicDetail, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) DescribeTopics([]string) ([]*sarama.TopicMetadata, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) DeleteTopic(string) error { return nil } + +func (f *FakeClusterAdmin) CreatePartitions(string, int32, [][]int32, bool) error { return nil } + +func (f *FakeClusterAdmin) AlterPartitionReassignments(string, [][]int32) error { return nil } + +func (f *FakeClusterAdmin) ListPartitionReassignments(string, []int32) (map[string]map[int32]*sarama.PartitionReplicaReassignmentsStatus, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) DeleteRecords(string, map[int32]int64) error { return nil } + +func (f *FakeClusterAdmin) DescribeConfig(sarama.ConfigResource) ([]sarama.ConfigEntry, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) AlterConfig(sarama.ConfigResourceType, string, map[string]*string, bool) error { + return nil +} + +func (f *FakeClusterAdmin) IncrementalAlterConfig(sarama.ConfigResourceType, string, map[string]sarama.IncrementalAlterConfigsEntry, bool) error { + return nil +} + +func (f *FakeClusterAdmin) CreateACL(sarama.Resource, sarama.Acl) error { return nil } + +func (f *FakeClusterAdmin) CreateACLs([]*sarama.ResourceAcls) error { return nil } + +func (f *FakeClusterAdmin) ListAcls(sarama.AclFilter) ([]sarama.ResourceAcls, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) DeleteACL(sarama.AclFilter, bool) ([]sarama.MatchingAcl, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) ListConsumerGroups() (map[string]string, error) { return nil, nil } + +func (f *FakeClusterAdmin) DescribeConsumerGroups([]string) ([]*sarama.GroupDescription, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) ListConsumerGroupOffsets(string, map[string][]int32) (*sarama.OffsetFetchResponse, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) DeleteConsumerGroupOffset(string, string, int32) error { return nil } + +func (f *FakeClusterAdmin) DeleteConsumerGroup(string) error { return nil } + +func (f *FakeClusterAdmin) DescribeCluster() ([]*sarama.Broker, int32, error) { return nil, 0, nil } + +func (f *FakeClusterAdmin) DescribeLogDirs([]int32) (map[int32][]sarama.DescribeLogDirsResponseDirMetadata, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) DescribeUserScramCredentials([]string) ([]*sarama.DescribeUserScramCredentialsResult, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) DeleteUserScramCredentials([]sarama.AlterUserScramCredentialsDelete) ([]*sarama.AlterUserScramCredentialsResult, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) UpsertUserScramCredentials([]sarama.AlterUserScramCredentialsUpsert) ([]*sarama.AlterUserScramCredentialsResult, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) DescribeClientQuotas([]sarama.QuotaFilterComponent, bool) ([]sarama.DescribeClientQuotasEntry, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) AlterClientQuotas([]sarama.QuotaEntityComponent, sarama.ClientQuotasOp, bool) error { + return nil +} + +func (f *FakeClusterAdmin) ElectLeaders(sarama.ElectionType, map[string][]int32) (map[string]map[int32]*sarama.PartitionResult, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) Controller() (*sarama.Broker, error) { return nil, nil } + +func (f *FakeClusterAdmin) Coordinator(string) (*sarama.Broker, error) { return nil, nil } + +func (f *FakeClusterAdmin) RemoveMemberFromConsumerGroup(string, []string) (*sarama.LeaveGroupResponse, error) { + return nil, nil +} + +func (f *FakeClusterAdmin) Close() error { + return f.closeFn() +} diff --git a/common/component/kafka/topics.go b/common/component/kafka/topics.go new file mode 100644 index 0000000000..6ddb6f7605 --- /dev/null +++ b/common/component/kafka/topics.go @@ -0,0 +1,49 @@ +/* +Copyright 2026 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package kafka + +import ( + "errors" + "fmt" + + "github.com/IBM/sarama" +) + +// ensureTopic creates the topic with the configured number of partitions if +// numPartitions metadata was set. It is a no-op if numPartitions is unset, +// and idempotent if the topic already exists. +func (k *Kafka) ensureTopic(topic string) error { + if k.numPartitions <= 0 { + return nil + } + if _, done := k.ensuredTopics.Load(topic); done { + return nil + } + + clients, err := k.latestClients() + if err != nil || clients == nil || clients.admin == nil { + return fmt.Errorf("failed to get kafka admin client: %w", err) + } + + err = clients.admin.CreateTopic(topic, &sarama.TopicDetail{ + NumPartitions: k.numPartitions, + ReplicationFactor: k.replicationFactor, + }, false) + if err != nil && !errors.Is(err, sarama.ErrTopicAlreadyExists) { + return fmt.Errorf("failed to create topic %s: %w", topic, err) + } + + k.ensuredTopics.Store(topic, struct{}{}) + return nil +} diff --git a/common/component/kafka/topics_test.go b/common/component/kafka/topics_test.go new file mode 100644 index 0000000000..bcda799c43 --- /dev/null +++ b/common/component/kafka/topics_test.go @@ -0,0 +1,155 @@ +/* +Copyright 2026 The Dapr Authors +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + http://www.apache.org/licenses/LICENSE-2.0 +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package kafka + +import ( + "errors" + "testing" + + "github.com/IBM/sarama" + "github.com/stretchr/testify/require" + + "github.com/dapr/components-contrib/common/component/kafka/mocks" + "github.com/dapr/kit/logger" +) + +func TestEnsureTopic(t *testing.T) { + t.Run("no-op when numPartitions is 0", func(t *testing.T) { + k := &Kafka{ + logger: logger.NewLogger("kafka_test"), + numPartitions: 0, + } + err := k.ensureTopic("my-topic") + require.NoError(t, err) + }) + + t.Run("no-op when numPartitions is negative", func(t *testing.T) { + k := &Kafka{ + logger: logger.NewLogger("kafka_test"), + numPartitions: -1, + } + err := k.ensureTopic("my-topic") + require.NoError(t, err) + }) + + t.Run("calls CreateTopic with correct args", func(t *testing.T) { + var gotTopic string + var gotDetail *sarama.TopicDetail + var gotValidate bool + + admin := mocks.NewClusterAdmin().WithCreateTopicFn( + func(topic string, detail *sarama.TopicDetail, validateOnly bool) error { + gotTopic = topic + gotDetail = detail + gotValidate = validateOnly + return nil + }, + ) + + k := &Kafka{ + logger: logger.NewLogger("kafka_test"), + numPartitions: 6, + replicationFactor: 3, + clients: &clients{ + admin: admin, + }, + } + + err := k.ensureTopic("orders") + require.NoError(t, err) + require.Equal(t, "orders", gotTopic) + require.NotNil(t, gotDetail) + require.Equal(t, int32(6), gotDetail.NumPartitions) + require.Equal(t, int16(3), gotDetail.ReplicationFactor) + require.False(t, gotValidate) + }) + + t.Run("treats ErrTopicAlreadyExists as success", func(t *testing.T) { + admin := mocks.NewClusterAdmin().WithCreateTopicFn( + func(string, *sarama.TopicDetail, bool) error { + return sarama.ErrTopicAlreadyExists + }, + ) + + k := &Kafka{ + logger: logger.NewLogger("kafka_test"), + numPartitions: 3, + replicationFactor: 1, + clients: &clients{ + admin: admin, + }, + } + + err := k.ensureTopic("existing-topic") + require.NoError(t, err) + + // Verify it's cached so subsequent calls don't hit CreateTopic again. + _, loaded := k.ensuredTopics.Load("existing-topic") + require.True(t, loaded) + }) + + t.Run("returns error for other CreateTopic failures", func(t *testing.T) { + admin := mocks.NewClusterAdmin().WithCreateTopicFn( + func(string, *sarama.TopicDetail, bool) error { + return errors.New("broker unavailable") + }, + ) + + k := &Kafka{ + logger: logger.NewLogger("kafka_test"), + numPartitions: 3, + replicationFactor: 1, + clients: &clients{ + admin: admin, + }, + } + + err := k.ensureTopic("fail-topic") + require.Error(t, err) + require.Contains(t, err.Error(), "failed to create topic fail-topic") + require.Contains(t, err.Error(), "broker unavailable") + + // Should NOT be cached on failure. + _, loaded := k.ensuredTopics.Load("fail-topic") + require.False(t, loaded) + }) + + t.Run("idempotent after successful create", func(t *testing.T) { + callCount := 0 + admin := mocks.NewClusterAdmin().WithCreateTopicFn( + func(string, *sarama.TopicDetail, bool) error { + callCount++ + return nil + }, + ) + + k := &Kafka{ + logger: logger.NewLogger("kafka_test"), + numPartitions: 3, + replicationFactor: 1, + clients: &clients{ + admin: admin, + }, + } + + err := k.ensureTopic("cached-topic") + require.NoError(t, err) + require.Equal(t, 1, callCount) + + // Second call should be a no-op (cached). + err = k.ensureTopic("cached-topic") + require.NoError(t, err) + require.Equal(t, 1, callCount) // Still 1 - not called again. + }) +} From 4e2cb7ea7154b91446f998251f0e717b2600eec2 Mon Sep 17 00:00:00 2001 From: akshitguptaa Date: Wed, 8 Jul 2026 02:02:31 +0530 Subject: [PATCH 2/3] kafka: wire ensureTopic into publish and subscribe paths Signed-off-by: akshitguptaa --- common/component/kafka/clients.go | 10 ++++++++++ common/component/kafka/producer.go | 8 ++++++++ common/component/kafka/subscriber.go | 10 ++++++++++ 3 files changed, 28 insertions(+) diff --git a/common/component/kafka/clients.go b/common/component/kafka/clients.go index ff571f3b12..76fa3ce676 100644 --- a/common/component/kafka/clients.go +++ b/common/component/kafka/clients.go @@ -9,6 +9,7 @@ import ( type clients struct { consumerGroup sarama.ConsumerGroup producer sarama.SyncProducer + admin sarama.ClusterAdmin } func (k *Kafka) latestClients() (*clients, error) { @@ -64,6 +65,15 @@ func (k *Kafka) latestClients() (*clients, error) { consumerGroup: cg, producer: p, } + + if k.numPartitions > 0 { + admin, err := sarama.NewClusterAdmin(k.brokers, k.config) + if err != nil { + return nil, fmt.Errorf("failed to create kafka admin client: %w", err) + } + newStaticClients.admin = admin + } + k.clients = &newStaticClients return k.clients, nil } diff --git a/common/component/kafka/producer.go b/common/component/kafka/producer.go index 758fa9377f..ad084c364b 100644 --- a/common/component/kafka/producer.go +++ b/common/component/kafka/producer.go @@ -62,6 +62,10 @@ func GetSyncProducer(config sarama.Config, brokers []string, maxMessageBytes int // Publish message to Kafka cluster. func (k *Kafka) Publish(_ context.Context, topic string, data []byte, metadata map[string]string) error { + if err := k.ensureTopic(topic); err != nil { + return err + } + clients, err := k.latestClients() if err != nil || clients == nil { return fmt.Errorf("failed to get latest Kafka clients: %w", err) @@ -121,6 +125,10 @@ func (k *Kafka) Publish(_ context.Context, topic string, data []byte, metadata m } func (k *Kafka) BulkPublish(_ context.Context, topic string, entries []pubsub.BulkMessageEntry, metadata map[string]string) (pubsub.BulkPublishResponse, error) { + if err := k.ensureTopic(topic); err != nil { + return pubsub.NewBulkPublishResponse(entries, err), err + } + clients, err := k.latestClients() if err != nil || clients == nil { err = fmt.Errorf("failed to get latest Kafka clients: %w", err) diff --git a/common/component/kafka/subscriber.go b/common/component/kafka/subscriber.go index 460893556f..2c8a93ee4f 100644 --- a/common/component/kafka/subscriber.go +++ b/common/component/kafka/subscriber.go @@ -30,6 +30,16 @@ func (k *Kafka) Subscribe(ctx context.Context, handlerConfig SubscriptionHandler k.subscribeTopics[topic] = handlerConfig } + // Best-effort topic creation: Subscribe is async and does not return + // errors, so we log failures instead of propagating them. The broker's + // own auto-create or external provisioning will handle topics if this + // fails. + for _, topic := range topics { + if err := k.ensureTopic(topic); err != nil { + k.logger.Errorf("failed to ensure topic %s: %v", topic, err) + } + } + k.logger.Debugf("Subscribing to topic: %v", topics) k.reloadConsumerGroup() From cebce8e59fba0184eca8842979ef7f2d95ec1b6d Mon Sep 17 00:00:00 2001 From: akshitguptaa Date: Wed, 8 Jul 2026 02:02:49 +0530 Subject: [PATCH 3/3] kafka: document numPartitions in pubsub and bindings metadata Signed-off-by: akshitguptaa --- bindings/kafka/metadata.yaml | 14 ++++++++++++++ common/component/kafka/metadata.go | 2 +- pubsub/kafka/metadata.yaml | 14 ++++++++++++++ 3 files changed, 29 insertions(+), 1 deletion(-) diff --git a/bindings/kafka/metadata.yaml b/bindings/kafka/metadata.yaml index 79ddc7bd8c..29f44c8145 100644 --- a/bindings/kafka/metadata.yaml +++ b/bindings/kafka/metadata.yaml @@ -342,6 +342,20 @@ metadata: The maximum size in bytes allowed for a single Kafka message. example: '2048' default: '1024' + - name: numPartitions + type: number + description: | + If set to a value greater than 0, dapr will create the topic with + this many partitions if it does not already exist. If unset + (default), dapr does not create topics; this is left to the + Kafka broker's own auto-create behavior or external provisioning. + example: '3' + - name: replicationFactor + type: number + description: | + Replication factor used when numPartitions is set and dapr + creates the topic. Defaults to 1. + example: '3' - name: consumeRetryInterval type: duration description: | diff --git a/common/component/kafka/metadata.go b/common/component/kafka/metadata.go index 3a13711c1a..f1af9ced81 100644 --- a/common/component/kafka/metadata.go +++ b/common/component/kafka/metadata.go @@ -382,7 +382,7 @@ func (k *Kafka) getKafkaMetadata(meta map[string]string) (*KafkaMetadata, error) if m.NumPartitions < 0 { return nil, errors.New("kafka error: 'numPartitions' must be a non-negative number") } - if m.NumPartitions > 0 && strings.ToLower(m.AuthType) == awsIAMAuthType { + if m.NumPartitions > 0 && strings.EqualFold(m.AuthType, awsIAMAuthType) { return nil, errors.New("kafka error: 'numPartitions' auto-topic-creation is not supported with authType 'awsiam'") } if m.NumPartitions > 0 && m.ReplicationFactor <= 0 { diff --git a/pubsub/kafka/metadata.yaml b/pubsub/kafka/metadata.yaml index 1f1d298e48..43ec11a21d 100644 --- a/pubsub/kafka/metadata.yaml +++ b/pubsub/kafka/metadata.yaml @@ -303,6 +303,20 @@ metadata: The maximum size in bytes allowed for a single Kafka message. example: '2048' default: '1024' + - name: numPartitions + type: number + description: | + If set to a value greater than 0, dapr will create the topic with + this many partitions if it does not already exist. If unset + (default), dapr does not create topics; this is left to the + Kafka broker's own auto-create behavior or external provisioning. + example: '3' + - name: replicationFactor + type: number + description: | + Replication factor used when numPartitions is set and dapr + creates the topic. Defaults to 1. + example: '3' - name: consumeRetryInterval type: duration description: |