偏移量

src/main/java/com/dsk/flink/dsc/sync/SyncCustomerDataSource.java

@@ -53,11 +53,11 @@ public class SyncCustomerDataSource {
         //TODO 到时需要改这里，改成正式的消费组
         FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<String>(envProps.getKafka_topic(), new SimpleStringSchema(), EtlUtils.getKafkaConfig(envProps.getKafka_brokers(), EtlUtils.getKafkaGroup(envProps), envProps.getKafka_username(),envProps.getKafka_password()));
         //System.out.println(envProps.getKafka_topic());
+        kafkaConsumer.setStartFromEarliest();
         //偏移量
         if (StrUtil.isNotBlank(offsetTimestamp)) {
             kafkaConsumer.setStartFromTimestamp(Long.parseLong(offsetTimestamp));
         }
-        kafkaConsumer.setStartFromEarliest();
 
         SingleOutputStreamOperator<String> kafkaSource = env.addSource(kafkaConsumer)
                 .setParallelism(1)