[rocketmq-spark] ask a master for help, repeated consumption occurs when the program restarts

[kai23333]

Reference code : https://github.com/apache/rocketmq-externals/blob/master/rocketmq-spark/src/test/java/org/apache/rocketmq/spark/streaming/RocketMqUtilsTest.java

My code :

Consumer Strategy Have chosen lastest, This setting doesn't seem to work, Program restart consumes historical data,

How to solve this problem?

The complete code is as follows:

        try {

            Map<String, String> optionParams = new HashMap<>();
            optionParams.put(RocketMQConfig.NAME_SERVER_ADDR, nameSrvAddr);
            SparkConf sparkConf = new SparkConf().setAppName("JavaCustomReceiver").setMaster("local[*]");
            JavaStreamingContext sc = new JavaStreamingContext(sparkConf, new Duration(duration));

            List<String> topics = new ArrayList<>();
            if (StringUtils.hasText(topic)) {
                for (String s : topic.split(";")) {
                    topics.add(s);
                }
            }

            LocationStrategy locationStrategy = LocationStrategy.PreferConsistent();

            JavaInputDStream<MessageExt> stream = RocketMqUtils.createJavaMQPullStream(sc, groupId,
                    topics, ConsumerStrategy.lastest(), false, false, false, locationStrategy, optionParams);

            stream.foreachRDD(new VoidFunction<JavaRDD<MessageExt>>() {

                private static final long serialVersionUID = 1L;

                @Override
                public void call(JavaRDD<MessageExt> messageExtJavaRDD) throws Exception {

                    JavaRDD<GPSRDD> GPSRDDJavaRDD = messageExtJavaRDD.map(new Function<MessageExt, GPSRDD>() {

                        private static final long serialVersionUID = 1L;

                        @Override
                        public GPSRDD call(MessageExt messageExt) throws Exception {

                            GPSRDD gps = new GPSRDD();
                            String xxx = new String(messageExt.getBody());
                            System.out.println(xxx);
                            return gps;
                        }
                    });

                }
            });

            sc.start();

        } catch (Exception e) {
            e.printStackTrace();
        }

[RockteMQ-AI]

Issue Evaluation

Category: type/question

This is a known behavior in Spark streaming with RocketMQ — when the application restarts, if the offset was not committed before the crash, the consumer will re-read from the last committed offset, causing repeated consumption.

Root Cause: Spark Streaming commits offsets after processing. If the app crashes between consuming and committing, those messages are re-delivered on restart. This is by design for at-least-once semantics.

Mitigation:

• Ensure checkpointing is enabled (checkpointDirectory) so Spark can recover from the last checkpoint
• Use failedLimit and maxRetries configurations to control retry behavior
• For exactly-once semantics, implement idempotent consumers or use transactional processing

---

Automated evaluation by github-manager

[RockteMQ-AI]

Issue Evaluation

Category: type/bug | Status: Needs Info

Repeated consumption on restart in the Spark connector typically indicates that checkpoint offsets are not being committed atomically with the Spark checkpoint. Please provide:

• The Spark version and checkpoint configuration
• Whether you're using enableSparkCheckpoint in the connector options
• The offset commit mode (sync vs async)

---

Automated evaluation by github-manager