Kafka: NumPartitions Option Has No Effect on Kafka Topic Creation

[yinguanhua]

Summary

Report: Kafka Topic Partition Configuration Not Working in DotNetCore.CAP.Kafka

Steps to Reproduce

Title: TopicOptions.NumPartitions Configuration Has No Effect on Kafka Topic Creation

Component: DotNetCore.CAP.Kafka

Version: 10.0.0

Environment:

.NET Version: 8.0

DotNetCore.CAP Version: [10.0.0]

DotNetCore.CAP.Kafka Version: [10.0.0]

Operating System: [Windows]

Issue Description:
When configuring Kafka topics through TopicOptions.NumPartitions and TopicOptions.ReplicationFactor, these settings are not being applied during topic creation. The Kafka topics are created with default partition counts (usually 1) instead of the specified values.

Steps to Reproduce:
Configure CAP with Kafka using the following code:

services.AddCap(options =>
{
options.UseKafka(kafkaOptions =>
{
kafkaOptions.Servers = "localhost:9092";

    // These settings should create topics with 15 partitions and 2 replicas
    kafkaOptions.TopicOptions.NumPartitions = 15;
    kafkaOptions.TopicOptions.ReplicationFactor = 2;
    
    kafkaOptions.MainConfig["group.id"] = "test.group";
});

options.UseSqlServer("connection_string");
options.DefaultGroupName = "test.group";

});

Expected Behavior

No response

Actual Behavior

No response

Log Output

CAP Configuration

Transport Used

None

Storage Provider

None

Environment

No response

Additional Context

No response

[yang-xiaodong]

See code here.

Make sure your customer service starts first and make sure that your Topic is created by CAP and allow.auto.create.topics is not set.

[yinguanhua]

I'm sure my topic was generated by CAP. The code is as follows:

await _capPublisher.PublishAsync("test.kafka.exchange3", "test.rabbitmq.key");

Startup
`services.AddCap(options =>
{
options.UseKafka(kafkaOptions =>
{
var servers = _Configuration["Kafka:Servers"];
kafkaOptions.Servers = servers;
kafkaOptions.ConnectionPoolSize = 100;
kafkaOptions.MainConfig["allow.auto.create.topics"] = "true";

   var securityProtocol = "SASL_PLAINTEXT"; 
    var saslMechanism = "PLAIN";
    var username = _Configuration["Kafka:Username"];
    var password = _Configuration["Kafka:Password"];

    if (!string.IsNullOrWhiteSpace(username) && !string.IsNullOrWhiteSpace(password))
    {
        kafkaOptions.MainConfig["security.protocol"] = securityProtocol;
        kafkaOptions.MainConfig["sasl.mechanism"] = saslMechanism;
        kafkaOptions.MainConfig["sasl.username"] = username;
        kafkaOptions.MainConfig["sasl.password"] = password;
    }

    kafkaOptions.TopicOptions.NumPartitions = 15; 
    kafkaOptions.TopicOptions.ReplicationFactor = 2; 
});
options.UseSqlServer(_Configuration.GetConnectionString("ScheduleDB"));
options.UseDashboard();
options.DefaultGroupName = "gateway.queue.group";
options.FailedRetryCount = 3;
options.ConsumerThreadCount = 15;
options.SucceedMessageExpiredAfter = 24 * 3600 * 60;

});`

[yinguanhua]

Could you please provide an example of configuring Kafka partitions

[yang-xiaodong]

This is a bug due to the introduction of Version10.