[GitHub] [rocketmq-externals] kai23333 opened a new issue, #901: [rocketmq-spark] [rocketmq-spark] ask a master for help, repeated consumption occurs when the program restarts

[via GitHub]

kai23333 opened a new issue, #901:
URL: https://github.com/apache/rocketmq-externals/issues/901

   Reference code : 
https://github.com/apache/rocketmq-externals/blob/master/rocketmq-spark/src/test/java/org/apache/rocketmq/spark/streaming/RocketMqUtilsTest.java
   
   
![code](https://user-images.githubusercontent.com/38802273/218986039-807f5465-3d45-4f33-8b0a-97b2ea9b526e.png)
   
   My code : 
   
   
![image](https://user-images.githubusercontent.com/38802273/218986812-fe74ab87-87fd-4040-9116-d8220b3a1303.png)
   
   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();
           }
   ```
   
   


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: dev-unsubscr...@rocketmq.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
us...@infra.apache.org