flink 消费kafka 程序重启后，从原先的自动提交的点继续消费，earliest 不用再从开始消费
如果开启了checkpoint 以 checkpoint为准 ，enable.auto.commit 失效，
如果没有开启，则以enable.auto.commit 为准
flink1.14.0

Kafka | Apache Flink

flink dataStream 方式

KafkaSource<String> source = KafkaSource.<String>builder()
                .setBootstrapServers("broker")
                .setTopics("topic")
                .setGroupId("group_id")
                .setProperty("enable.auto.commit","true")
//                .setProperty("auto.offset.reset","earliest")
                .setProperty("auto.offset.reset","latest")
                .setProperty("auto.commit.interval.ms","1000")
//                .setStartingOffsets(OffsetsInitializer.earliest())
                .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        DataStreamSource<String> kafka_source = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
        kafka_source.print();

flink sql 方式

重点是这四个属性
'scan.startup.mode'='group-offsets', // 默认
'properties.enable.auto.commit' = 'true',
'properties.auto.offset.reset.strategy' = 'earliest',
'properties.auto.commit.interval.ms' = '1000'

CREATE TABLE  `kafka_order_stream` (
  `id` VARCHAR ,
  `name` VARCHAR 
) WITH (
  'format' = 'json',
  'json.ignore-parse-errors' = 'true',
  'json.fail-on-missing-field' = 'false',
  'properties.bootstrap.servers' = 'broker',
  'connector' = 'kafka',
  'topic' = 'topic',
  'scan.startup.mode'='group-offsets', // 默认
  'properties.group.id' = 'group_id',
  'properties.enable.auto.commit' = 'true',
  'properties.auto.offset.reset.strategy' = 'earliest',
  'properties.auto.commit.interval.ms' = '1000'
);