MassTransit向Confluent Cloud Kafka发送大消息遇大小限制异常

核心问题 #

你在MassTransit全局配置中设置的MessageMaxBytes并未正确传递到Kafka生产者/消费者的具体配置项中，且缺少了大消息场景下的关键配置参数，导致即使Confluent Cloud主题已允许2.5MB消息，仍无法发送超过1MB的内容。

具体解决方案 #

1. 修正生产者配置 #

需在生产者专属配置中设置消息大小限制，而非全局kafkaFactory层面。修改后的生产者代码如下：

services.AddMassTransit(x =>
{
    x.UsingInMemory((context, cfg) =>
    {
        cfg.ConfigureEndpoints(context);
    });
    
    x.AddRider(rider =>
    {
        // 在添加生产者时单独配置ProducerConfig
        rider.AddProducer<[...Event...]>(topicName: "<TOPIC NAME...>", (context, producerConfig) =>
        {
            producerConfig.MessageMaxBytes = 2500000; // 单条消息最大字节数
            producerConfig.MaxRequestSize = 2500000; // 单个请求的最大字节数，需大于等于MessageMaxBytes
        });
        
        rider.UsingKafka((riderContext, kafkaFactory) => 
        {
            kafkaFactory.SecurityProtocol = SecurityProtocol.SaslSsl;
            kafkaFactory.Host(builder.Configuration["ConfluentCloud:BootstrapServerHost"], configureHost =>
            {
                configureHost.UseSasl(saslConfig =>
                {
                    saslConfig.Mechanism = SaslMechanism.Plain;
                    saslConfig.Username = builder.Configuration["ConfluentCloud:API_KEY"];
                    saslConfig.Password = builder.Configuration["ConfluentCloud:API_SECRET"];
                });
            });
        });
    });
});

2. 修正消费者配置 #

消费者也需配置对应参数以支持接收大消息，修改后的消费者代码如下：

services.AddMassTransit(x =>
{
    x.UsingInMemory((context, cfg) =>
    {
        cfg.ConfigureEndpoints(context);
    });

    x.AddRider(rider =>
    {
        rider.AddConsumer<[...TOPIC_CONSUMER...]>();

        rider.UsingKafka((riderContext, kafkaFactory) =>
        {
            kafkaFactory.SecurityProtocol = SecurityProtocol.SaslSsl;
            kafkaFactory.Host(builder.Configuration["ConfluentCloud:BootstrapServerHost"], configureHost =>
            {
                configureHost.UseSasl(saslConfig =>
                {
                    saslConfig.Mechanism = SaslMechanism.Plain;
                    saslConfig.Username = builder.Configuration["ConfluentCloud:API_KEY"];
                    saslConfig.Password = builder.Configuration["ConfluentCloud:API_SECRET"];
                });
            });

            var consumerConfig = new ConsumerConfig()
            {
                GroupId = "<GROUPID>",
                AutoOffsetReset = AutoOffsetReset.Latest,
                EnableAutoCommit = false,
                // 添加消费者端大消息配置
                FetchMaxBytes = 2500000, // 单次拉取的最大字节数
                MaxPartitionFetchBytes = 2500000 // 每个分区单次拉取的最大字节数
            };

            kafkaFactory.TopicEndpoint<[...Event...]>(
                topicName: "<TOPIC NAME>", consumerConfig, kafkaTopicReceiveEndpointConfig =>
                {
                    kafkaTopicReceiveEndpointConfig.ConfigureConsumer<[...TOPIC_CONSUMER...]>(riderContext);
                });
        });
    });
});

3. 额外验证点 #

• 若你的事件模型中图片以Base64字符串存储，Base64会使原始字节体积膨胀约33%，2MB原始图片序列化后会达到约2.66MB，超过2.5MB限制。建议直接用字节数组存储图片数据，或把消息大小限制提高到3MB以上。

内容的提问来源于stack exchange，提问作者user20291437