如何在Flink作业执行期间记录未捕获异常以对接Sentry？

我之前在给Flink集群接入Sentry时也碰到过一模一样的需求——毕竟默认只抓WARN及以上日志，得确保所有未捕获异常都能被Sentry精准捕获到。下面给你几个实用的方案：

方案一：在算子中手动捕获并记录异常（推荐） #

对于自定义算子，用RichFunction作为基类，把核心业务逻辑包裹在try-catch块里，捕获到异常后直接打ERROR级别日志，同时可以选择重新抛出异常让Flink的重启策略生效。这样既能让Sentry捕获到异常日志，也不影响Flink本身的容错机制。

示例代码：

import org.apache.flink.api.common.functions.RichMapFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class ErrorHandlingMapFunction extends RichMapFunction<String, String> {
    private static final Logger LOG = LoggerFactory.getLogger(ErrorHandlingMapFunction.class);

    @Override
    public String map(String input) throws Exception {
        try {
            // 这里写你的业务处理逻辑
            return processInput(input);
        } catch (Exception e) {
            // 记录ERROR日志，Sentry会自动捕获这条日志及异常栈
            LOG.error("算子处理数据时发生未捕获异常，输入数据：{}", input, e);
            // 重新抛出异常，让Flink的重启策略触发
            throw e;
        }
    }

    private String processInput(String input) {
        // 模拟业务逻辑，可能抛出异常
        if (input == null || input.isEmpty()) {
            throw new IllegalArgumentException("输入数据不能为空");
        }
        return input.toUpperCase();
    }
}

方案二：通过日志框架捕获Flink内置的异常日志 #

如果不想逐个修改算子，可以利用Flink本身的日志输出机制。当算子抛出未捕获异常时，Flink的TaskManager会在org.apache.flink.runtime.taskmanager.Task这个类下打印ERROR级别的日志，包含完整的异常栈信息。你只需要在日志配置（比如Logback/Log4j）中确保这个类的ERROR日志能被Sentry的Appender捕获。

以Logback为例，修改logback.xml：

<!-- 配置Sentry Appender -->
<appender name="SENTRY" class="io.sentry.logback.SentryAppender">
    <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
        <level>WARN</level>
    </filter>
    <!-- 这里可以添加Sentry的DSN等配置 -->
</appender>

<!-- 让Task类的ERROR日志输出到Sentry -->
<logger name="org.apache.flink.runtime.taskmanager.Task" level="ERROR" additivity="false">
    <appender-ref ref="SENTRY"/>
</logger>

这样所有算子未捕获的异常都会被TaskManager记录下来，进而被Sentry捕获。

方案三：处理作业执行的最终异常 #

正如你提到的，当重启策略失效后，execute()方法会抛出最终异常，这时候只需要在主函数里用try-catch包裹execute()调用，记录ERROR日志即可：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class MyFlinkJob {
    private static final Logger LOG = LoggerFactory.getLogger(MyFlinkJob.class);

    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 配置作业、添加算子...
        env.addSource(new MySource())
           .map(new ErrorHandlingMapFunction())
           .addSink(new MySink());

        try {
            env.execute("My Flink Job");
        } catch (Exception e) {
            LOG.error("作业执行失败，触发最终异常", e);
            // 退出程序，避免进程挂起
            System.exit(1);
        }
    }
}

小提示 #

• 如果用方案一，建议把异常处理逻辑封装成工具类，避免重复代码；
• 确保Sentry的日志Appender配置正确，能捕获到WARN及以上级别的日志；
• 测试时可以故意抛出异常，验证Sentry是否能收到对应的日志和异常信息。

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