Spark Streaming消费NATS JetStream重复输出历史消息的解决方案咨询

使用nats-spark-connector连接NATS JetStream，通过Spark Java代码消费消息时，出现每个批次都会输出已处理过的历史消息的问题，即使设置outputMode("append")或outputMode("update")也无法实现仅接收新消息的预期效果。

现有代码与操作步骤 #

Spark Java消费代码 #

private static void sparkNatsTester() {
        SparkSession spark = SparkSession.builder()
                .appName("spark-with-nats")
                .master("local")
//              .config("spark.logConf", "false")
                  .config("spark.jars",
                  "libs/nats-spark-connector-balanced_2.12-1.1.4.jar,"+"libs/jnats-2.17.1.jar"
                  )
//                .config("spark.executor.instances", "2")
//                .config("spark.cores.max", "4")
//                .config("spark.executor.memory", "2g")
                  .getOrCreate();
        System.out.println("sparkSession : "+ spark);
        Dataset<Row> df = spark.readStream()
                .format("nats")
                .option("nats.host", "localhost")
                .option("nats.port", 4222)
                .option("nats.stream.name", "my_stream3")
                .option("nats.stream.subjects", "mysub3")
                // wait 90 seconds for an ack before resending a message
                .option("nats.msg.ack.wait.secs", 90)
                //.option("nats.num.listeners", 2)
                // Each listener will fetch 10 messages at a time
               // .option("nats.msg.fetch.batch.size", 10)
                .load();
        System.out.println("Successfully read nats stream !");
        
        StreamingQuery query;
        try {
            query = df.writeStream()
                      .outputMode("append")
                      .format("console")
                      .option("truncate", false)
                      .start();
            query.awaitTermination();
        } catch (Exception e) {
            e.printStackTrace();
        } 
    }

消息发布命令 #

nats pub foo --count=1000 --sleep 1s "publication #{{Count}} @ {{TimeStamp}}" 

问题分析 #

1. 消费位置未持久化：当前配置未启用JetStream的持久化消费者，导致每次Spark流处理启动或批次运行时，都会从头开始消费流中的所有消息，而非从上次停止的位置继续。
2. 主题不匹配：发布命令使用的主题是foo，但代码中订阅的流绑定主题是mysub3，新发布的消息不会进入目标流，导致消费端一直读取历史消息。

解决方案 #

1. 修正消息发布主题 #

将发布命令改为目标主题mysub3，确保消息进入指定流：

nats pub mysub3 --count=1000 --sleep 1s "publication #{{Count}} @ {{TimeStamp}}" 

2. 添加持久化消费者配置 #

在Spark的readStream配置中添加以下选项，启用持久化消费并指定起始位置：

Dataset<Row> df = spark.readStream()
        .format("nats")
        .option("nats.host", "localhost")
        .option("nats.port", 4222)
        .option("nats.stream.name", "my_stream3")
        .option("nats.stream.subjects", "mysub3")
        .option("nats.msg.ack.wait.secs", 90)
        // 配置持久化消费者名称，NATS会保存该消费者的消费进度
        .option("nats.consumer.durable", "spark-durable-consumer")
        // 设置消费起始位置：last表示从上次消费的位置继续，new表示只消费订阅后新产生的消息
        .option("nats.start.position", "last")
        .load();

3. 验证流配置 #

确认NATS JetStream的my_stream3流确实绑定了mysub3主题，可通过NATS命令行工具查看流信息：

nats stream info my_stream3

修改后重启Spark消费程序，新发布的消息会被单独消费，不会重复输出历史消息。

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