Kafka: add numPartitions and replicationFactor metadata for auto topic creation

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: |

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
 	}

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()
+			}
 		}
 	}
 

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.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 {
+		m.ReplicationFactor = 1
+	}
+
 	// confirm client connection fields are valid
 	if m.ClientConnectionTopicMetadataRefreshInterval <= 0 {
 		m.ClientConnectionTopicMetadataRefreshInterval = defaultClientConnectionTopicMetadataRefreshInterval

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()

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()
+}

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)

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()