如何避免Azure Function停止消费Kafka Topic消息？

你的Azure Function Kafka消费者停止消费的核心原因，大概率是未处理的异常导致Kafka触发器宿主进程终止或消费者实例崩溃，尽管你在ProccessMessage中添加了异常捕获，但仍存在几个关键漏洞：

潜在问题点 #

1. 日志保存操作未捕获异常
   在ProccessMessage的catch块中，调用await logEntry.Save(connection)时如果抛出异常（比如数据库连接中断、写入失败），该异常会直接逃出catch块，成为Function方法的未处理异常，最终导致Kafka消费者停止工作。

2. 数据库连接复用风险
   你在单个Function执行中复用同一个SqlConnection处理批量消息，若某条消息的操作导致连接进入异常状态（如死锁、连接超时），后续所有消息的处理都会失败，甚至引发连锁异常。

3. Function方法无外层异常捕获
   Consumer1和Consumer2方法本身没有外层try-catch，任何逃出ProccessMessage的异常都会直接传递给Azure Function宿主，触发Kafka触发器的终止逻辑。

4. SqlConnection未显式打开
   创建SqlConnection后未调用OpenAsync()，部分数据库操作（如Save）可能隐含打开连接，但如果连接池资源不足，会抛出未预期的异常。

修复方案 #

1. 确保日志操作的异常被捕获 #

在ProccessMessage的catch块内，对日志保存操作添加二次异常捕获，避免异常扩散：

private async Task ProccessMessage(string msgString, ILogger log)
{
    var logEntry = new Log
    {
        id = Guid.NewGuid().ToString(),
        error = string.Empty,
        input = msgString,
        updated_at = DateTime.UtcNow
    };

    // 为每条消息单独创建连接，利用连接池管理
    await using (var connection = new SqlConnection(connectionString))
    {
        await connection.OpenAsync();
        try
        {
            var input = JsonConvert.DeserializeObject<Message>(msgString);
            var entity = GetEntity(input.Metadata.TableName, msgString);
            var operation = input.Metadata.Operation;
            await entity.Save(connection, operation);
            await logEntry.Save(connection);
        }
        catch (Exception e)
        {
            logEntry.error = e.ToString();
            try
            {
                await logEntry.Save(connection);
            }
            catch (Exception logEx)
            {
                // 日志保存失败时，记录到Function内置日志
                log.LogError(logEx, "Failed to save error log for message: {MsgContent}", msgString);
            }
        }
    }
}

2. 添加Function方法级别的异常捕获 #

在Consumer1和Consumer2中添加外层try-catch，捕获所有未处理异常并记录，防止异常传递给宿主：

[FunctionName(nameof(Consumer1))]
public async Task Consumer1(
    [KafkaTrigger("BrokerList",
        "kafka-default-topic",
        Username = "kafka",
        Password = "%KafkaPassword%",
        Protocol = BrokerProtocol.SaslSsl,
        AuthenticationMode = BrokerAuthenticationMode.ScramSha512,
        ConsumerGroup = "foo")] KafkaEventData<string>[] events,
    ILogger log)
{
    try
    {
        foreach (KafkaEventData<string> eventData in events)
        {
            await ProccessMessage(eventData.Value, log);
        }
    }
    catch (Exception e)
    {
        log.LogError(e, "Unhandled exception in Consumer1 processing batch");
        // 此处无需重新抛出，捕获后宿主不会终止，Kafka触发器会重试当前批次
    }
}

3. 优化数据库连接使用 #

将数据库连接的创建移到ProccessMessage内部，为每条消息单独管理连接，利用SQL Server连接池自动复用连接，避免单条消息的连接问题影响批量处理。

4. 使用配置绑定获取连接字符串 #

避免硬编码连接字符串，改用环境变量获取，符合Azure Function最佳实践：

private readonly string connectionString;

public Consumers(IConfiguration configuration)
{
    connectionString = configuration.GetConnectionString("DbConnection");
}

5. 可选：添加瞬时异常重试策略 #

使用Polly库对数据库操作添加重试逻辑，处理网络波动、连接超时等瞬时异常：

// 定义重试策略
private static readonly AsyncRetryPolicy retryPolicy = Policy
    .Handle<SqlException>(ex => ex.IsTransient())
    .Or<TimeoutException>()
    .WaitAndRetryAsync(3, retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)));

// 在Save操作中使用
await retryPolicy.ExecuteAsync(async () => await entity.Save(connection, operation));

额外排查步骤 #

• 查看Azure Function的Application Insights日志，搜索Unhandled exception或Kafka相关错误，确认具体异常信息。
• 检查Kafka集群的消费者组状态，确认是否存在偏移量提交失败或消费者实例崩溃的情况。
• 调整KafkaTrigger的MaxBatchSize参数，减小批量处理的消息数量，降低单批次异常的影响范围。

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