Spark Streaming对接Kafka：如何自动处理Driver端异常实现无间断运行？

首先咱们理清问题根源：当其他写入Kafka的应用重启时，Kafka集群会触发分区Leader重新选举，这期间Spark Streaming的Direct Stream拉取元数据或数据时，可能拿到异常的分区偏移量信息，进而导致计算numRecords为负数，抛出IllegalArgumentException。要实现无人干预的自动恢复，咱们可以从异常捕获重启和参数优化两方面入手，以下是具体的Java实现方案：

一、通过循环重启机制实现自动恢复 #

最直接的方式是在主程序中用循环包裹StreamingContext的初始化和启动逻辑，当捕获到目标异常时，优雅停止现有上下文并重新初始化：

import org.apache.spark.SparkConf;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

public class AutoRestartSparkStreaming {
    private static final String KAFKA_BOOTSTRAP_SERVERS = "your-kafka-brokers:9092";
    private static final String TOPIC = "your-topic";
    private static final String GROUP_ID = "your-group-id";

    public static void main(String[] args) {
        while (true) {
            JavaStreamingContext jssc = null;
            try {
                // 初始化Spark配置
                SparkConf conf = new SparkConf()
                        .setAppName("AutoRestartKafkaStream")
                        .setMaster("local[*]"); // 生产环境替换为集群模式

                // 初始化StreamingContext，批次间隔根据业务调整
                jssc = new JavaStreamingContext(conf, Durations.seconds(5));

                // 配置Kafka消费者参数
                Map<String, Object> kafkaParams = new HashMap<>();
                kafkaParams.put("bootstrap.servers", KAFKA_BOOTSTRAP_SERVERS);
                kafkaParams.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
                kafkaParams.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
                kafkaParams.put("group.id", GROUP_ID);
                kafkaParams.put("auto.offset.reset", "latest");
                kafkaParams.put("enable.auto.commit", false);
                // 关键参数：缩短元数据刷新间隔，快速感知Leader变化
                kafkaParams.put("metadata.max.age.ms", "30000");

                // 创建Direct Stream
                var stream = KafkaUtils.createDirectStream(
                        jssc,
                        LocationStrategies.PreferConsistent(),
                        ConsumerStrategies.Subscribe(Collections.singletonList(TOPIC), kafkaParams)
                );

                // 你的业务处理逻辑
                stream.foreachRDD(rdd -> {
                    // 处理RDD数据
                    rdd.foreach(record -> {
                        System.out.println("Received: " + record.value());
                    });
                    // 手动提交偏移量（可选，根据业务需求）
                    // stream.asInstanceOf[CanCommitOffsets].commitAsync(rdd.asInstanceOf[HasOffsetRanges].offsetRanges());
                });

                // 启动StreamingContext
                jssc.start();
                jssc.awaitTermination();
            } catch (Exception e) {
                // 捕获目标异常：numRecords must not be negative
                if (e.getMessage() != null && e.getMessage().contains("numRecords must not be negative")) {
                    System.err.println("触发自动重启：" + e.getMessage());
                } else {
                    // 其他异常也可以考虑重启，根据业务调整
                    System.err.println("发生异常，准备重启：" + e.getMessage());
                }
                // 优雅停止现有上下文
                if (jssc != null) {
                    try {
                        jssc.stop(true, true); // 停止上下文并关闭SparkContext
                    } catch (Exception stopEx) {
                        System.err.println("停止上下文时出错：" + stopEx.getMessage());
                    }
                }
                // 重启前加短暂延迟，避免频繁重启
                try {
                    Thread.sleep(5000);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                }
            }
        }
    }
}

二、使用StreamingListener监听异常并重启 #

另一种更灵活的方式是通过StreamingListener监听JobScheduler的错误事件，当捕获到目标异常时触发重启：

import org.apache.spark.streaming.scheduler.StreamingListener;
import org.apache.spark.streaming.scheduler.StreamingListenerError;

public class RestartStreamingListener implements StreamingListener {
    private final JavaStreamingContext jssc;

    public RestartStreamingListener(JavaStreamingContext jssc) {
        this.jssc = jssc;
    }

    @Override
    public void onStreamingListenerError(StreamingListenerError error) {
        Throwable cause = error.exception();
        // 检查是否是目标异常
        if (cause instanceof IllegalArgumentException && cause.getMessage().contains("numRecords must not be negative")) {
            System.err.println("监听到目标异常，准备重启StreamingContext...");
            // 启动新线程执行重启，避免阻塞Listener线程
            new Thread(() -> {
                try {
                    // 停止当前上下文
                    jssc.stop(true, true);
                    // 重新初始化并启动（这里可以封装成初始化方法，避免重复代码）
                    restartStreamingContext();
                } catch (Exception e) {
                    System.err.println("重启失败：" + e.getMessage());
                }
            }).start();
        }
    }

    private void restartStreamingContext() {
        // 复用初始化逻辑，创建新的JavaStreamingContext并启动
        SparkConf conf = new SparkConf()
                .setAppName("AutoRestartKafkaStream")
                .setMaster("local[*]");
        JavaStreamingContext newJssc = new JavaStreamingContext(conf, Durations.seconds(5));
        // 重新配置Kafka参数和DStream...
        // 添加当前Listener到新的上下文
        newJssc.addStreamingListener(new RestartStreamingListener(newJssc));
        newJssc.start();
        try {
            newJssc.awaitTermination();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

然后在主程序中添加这个Listener：

public static void main(String[] args) {
    SparkConf conf = new SparkConf()
            .setAppName("AutoRestartKafkaStream")
            .setMaster("local[*]");
    JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
    // 添加重启Listener
    jssc.addStreamingListener(new RestartStreamingListener(jssc));
    // 配置Kafka Stream和业务逻辑...
    jssc.start();
    try {
        jssc.awaitTermination();
    } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
    }
}

三、优化Kafka参数减少异常触发概率 #

除了重启机制，调整以下Kafka消费者参数可以降低Leader不可用带来的影响：

• metadata.max.age.ms：默认300000ms（5分钟），调小到30000ms（30秒），让Spark更快刷新Kafka元数据，及时感知Leader变化。
• fetch.max.wait.ms：默认500ms，保持默认或适当调小，避免长时间等待元数据。
• retries：设置重试次数，比如3，让消费者在遇到Leader不可用时自动重试几次，减少异常抛出概率。

注意事项 #

1. 偏移量管理：如果使用手动提交偏移量，重启时要确保偏移量已经正确提交到Kafka或外部存储（比如ZooKeeper、Redis），避免重复消费或数据丢失。
2. 资源释放：重启前必须优雅停止现有StreamingContext，避免资源泄漏。
3. 避免无限重启：可以添加重启次数限制，防止因其他致命错误导致无限循环重启。

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