更新kafka消费的策略

iot-machine-state-event-generator-job/src/main/java/com/qniao/iot/machine/event/generator/job/IotMachineEventGeneratorJob.java

@@ -66,12 +66,11 @@ public class IotMachineEventGeneratorJob {
         env.enableCheckpointing(60000L, CheckpointingMode.EXACTLY_ONCE);
         // 设置并行度为1，并行度要小于等于kafka topic的分区数，否则其他并行度分配不到数据
         // env.setParallelism(1);
-        TopicPartition topicPartition = new TopicPartition(ApolloConfig.get(ConfigConstant.SOURCE_KAFKA_TOPICS), 0);
         KafkaSource<MachineIotDataReceivedEvent> source = KafkaSource.<MachineIotDataReceivedEvent>builder()
                 .setBootstrapServers(ApolloConfig.get(ConfigConstant.SOURCE_KAFKA_BOOTSTRAP_SERVERS))
                 .setTopics(ApolloConfig.get(ConfigConstant.SOURCE_KAFKA_TOPICS))
                 .setGroupId(ApolloConfig.get(ConfigConstant.SOURCE_KAFKA_GROUP_ID))
-                .setStartingOffsets(OffsetsInitializer.latest())
+                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
                 .setProperty(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "8000")
                 .setValueOnlyDeserializer(new MachineIotDataReceivedEventKafkaDeserializationSchema())
                 .build();

iot-machine-state-event-generator-job/src/main/resources/META-INF/app.properties

@@ -3,4 +3,4 @@ app.id=machine-state-event-generator
 
 # ???? 8.135.8.221
 # ???? 47.112.164.224
-apollo.meta=http://8.135.8.221:5000
+apollo.meta=http://47.112.164.224:5000