diff --git a/iot-machine-state-event-generator-job/src/main/java/com/qniao/iot/machine/event/generator/job/IotMachineEventGeneratorJob.java b/iot-machine-state-event-generator-job/src/main/java/com/qniao/iot/machine/event/generator/job/IotMachineEventGeneratorJob.java index 71868b9..b42e97c 100644 --- a/iot-machine-state-event-generator-job/src/main/java/com/qniao/iot/machine/event/generator/job/IotMachineEventGeneratorJob.java +++ b/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 source = KafkaSource.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(); diff --git a/iot-machine-state-event-generator-job/src/main/resources/META-INF/app.properties b/iot-machine-state-event-generator-job/src/main/resources/META-INF/app.properties index ed4c09b..01661b3 100644 --- a/iot-machine-state-event-generator-job/src/main/resources/META-INF/app.properties +++ b/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 \ No newline at end of file +apollo.meta=http://47.112.164.224:5000 \ No newline at end of file