flink-1.16 table sql 消费 kafka 数据，指定时间戳位置消费数据报错：Invalid negative offset 问题解决

1 背景

1.使用 flink-1.16 的 table sql 消费 kafka数据，并使用 sql 计算指标，然后写入 doris；

2.指标计算时，需要统计当日数据条数，考虑到作业异常退出被重新拉起时，需要从零点开始消费，所以指定 'scan.startup.mode' = 'timestamp'，'scan.startup.timestamp-millis' = '当日零点时间戳' 方式创建 kafka table：

s"""
  |CREATE TABLE qysfxjKafkaTable (
  |xlid STRING,
  |available_status STRING,
  |sendtime STRING,
  |`ts` TIMESTAMP(3) METADATA FROM 'timestamp'
  |) WITH (
  |'connector' = 'kafka',
  |'topic' = '${param.getProperty("qysfxjTopic")}',
  |'scan.startup.mode' = 'timestamp'，'scan.startup.timestamp-millis' = '当日零点时间戳'
  |'properties.group.id' = '${param.getProperty("qysfxjTopicGroupId")}',
  |'properties.bootstrap.servers' = '${param.getProperty("brokers")}',
  |'properties.auto.offset.reset' = 'earliest',
  |'json.ignore-parse-errors' = 'true',
  |'json.fail-on-missing-field' = 'false',
  |'format' = 'json')
  |""".stripMargin

3.启动时报 kakfa 的错误，Invalid negative offset，即 flink 使用了一个不正确的 offset 到 kafka 消费数据，经过排查 topic 中最新一条数据的时间，在今日零点之前，也就是说，kafka table sql 中指定今日零点的时间戳，落后于 kafka 最新数据的时间；

2 解决方案

1.两种解决方案，① 从检查点启动作业；② 根据 kafka 数据时间，调整消费的时间；考虑到第一次启动可能 topic 也没有数据，且如果检查点失败会导致作业无法从检查点恢复的情况，决定采用 ② 方案解决；

2.方案步骤

1.使用 kafka java api，获取 topic 中最后一条数据，根据数据的时间戳初始化创建 kafka table sql 的启动时间；

2.获取到 kafka 最后一条数据的场景有两种：① kafka 中最新一条数据时间早于零点(报错的场景)；② kafka 中最新一条数据时间晚于零点；

3.根据以上步骤，实现代码，代码会返回一个时间戳，0或者最后一条数据时间戳：

def getTopicLatestRecordTimeStamp(brokers: String,topic: String): Long ={
    var retMillis = 0L
    val props = new Properties();
    props.put("bootstrap.servers", brokers);
    props.put("group.id", "test");
    props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
    props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

    val consumer = new KafkaConsumer[String,String](props);
    try {
      // 订阅 topic
      consumer.subscribe(Collections.singletonList(topic))
      // 获取并订阅全部分区
      var assigment = consumer.assignment()
      while (assigment.size() == 0) {
        consumer.poll(Duration.ofMillis(1000L))
        assigment = consumer.assignment()
      }
      println(assigment)
      val earliestOffset = consumer.beginningOffsets(assigment)
      val lastOffset = consumer.endOffsets(assigment)
      println("assigment: " + assigment + ",topic earliestOffset:" + earliestOffset + ",topic endOffsets:" + lastOffset)
      // 遍历所有分区，获取最新的 offset
      val lastOffsetIter = lastOffset.entrySet().iterator()
      while (lastOffsetIter.hasNext){
        val ele = lastOffsetIter.next()
        if(ele.getValue != 0L){
          // 分区有数据
          consumer.seek(ele.getKey, ele.getValue - 1)
          val records = consumer.poll(1000).iterator()
          while (records.hasNext){
            val next = records.next()
            if(next.timestamp() > retMillis){
              retMillis = next.timestamp()
            }
            System.out.println("Timestamp of last record: " + next.timestamp())
          }
        }
      }
      retMillis
    } finally {
      consumer.close();
    }
  }

4.根据返回的时间戳，于当日零点判断，如果返回的时间戳早于零点，使用 latest-offset，返回的时间戳晚于当日零点，使用零点启动即可，以下代码返回使用的是时间戳启动，还是 latest-offset 启动：

if(parameterTool.get("qysfxjTopicStartFrom","latest").equals("latest")){
      val toAssignmentTime = TimeHandler.getMidNightMillions()
      val latestTime = KfkUtil.getTopicLatestRecordTimeStamp(pro.get("brokers").toString,pro.get("qysfxjTopic").toString)
      // 如果最后一条数据在 toAssignmentTime 之前，则使用 latest-offset 启动消费
      if(toAssignmentTime > latestTime){
        pro.put("qysfxjTopicStartFrom","latest-offset")
      }else {
        pro.put("qysfxjTopicStartFrom",(toAssignmentTime).toString)
      }
    }else {
      pro.put("qysfxjTopicStartFrom",parameterTool.get("qysfxjTopicStartFrom"))
    }

5.根据时间戳，还是 latest-offset，生成 sql 中的 scan 片段：

/**
   * 根据 startFrom，判断是从什么位置消费。
   *
   * @param startFrom：earliest-offset，latest-offset，group-offsets，timestamp
   * @return
   */
  def getKafkaSQLScanStr(startFrom: String): String = {
    var scanStartup = ""
    if(Character.isDigit(startFrom.trim()(0))){
      scanStartup =
        "'scan.startup.mode' = 'timestamp'," +
          s"'scan.startup.timestamp-millis' = '${startFrom.trim()}',"
    }else {
      scanStartup =
        s"'scan.startup.mode' = '${startFrom}',"
    }
    scanStartup
  }

6.完整table sql 拼接：

val qysfxjKafkaSource =
      s"""
        |CREATE TABLE qysfxjKafkaTable (
        |xlid STRING,
        |available_status STRING,
        |sendtime STRING,
        |`ts` TIMESTAMP(3) METADATA FROM 'timestamp'
        |) WITH (
        |'connector' = 'kafka',
        |'topic' = '${param.getProperty("qysfxjTopic")}',
        |${TXGJUtils.getKafkaSQLScanStr(qysfxjTopicStartFrom)}
        |'properties.group.id' = '${param.getProperty("qysfxjTopicGroupId")}',
        |'properties.bootstrap.servers' = '${param.getProperty("brokers")}',
        |'properties.auto.offset.reset' = 'earliest',
        |'json.ignore-parse-errors' = 'true',
        |'json.fail-on-missing-field' = 'false',
        |'format' = 'json')
        |""".stripMargin