public void recordLatency(StreamRecord<RowData> element) {
    if (latencyEnable) {
      if (findColumn) {
        RowData rowData = element.getValue();
        if (rowData.getRowKind() == RowKind.UPDATE_BEFORE
            || rowData.getRowKind() == RowKind.DELETE) {
          return;
        }

        Object value = modifyTimeGetter.getFieldOrNull(rowData);
        if (value == null) {
          return;
        }
        if (value instanceof LocalDateTime) {
          LocalDateTime localDateTime = (LocalDateTime) value;
          long eventTime = localDateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
          this.currentLatency = System.currentTimeMillis() - eventTime;
        } else if (value instanceof Long) {
          this.currentLatency = System.currentTimeMillis() - (Long) value;
        } else {
          LOG.warn("eventTimeColumn is not LocalDateTime/Long, {}", value.getClass());
        }
      } else if (element.hasTimestamp()) {
        this.currentLatency = System.currentTimeMillis() - element.getTimestamp();
      }
    }
  }
A generator that generates the latency metrics of the writing operators in flink applications.
amoro-format-mixed/amoro-mixed-flink/amoro-mixed-flink-common/src/main/java/org/apache/amoro/flink/metric/MetricsGenerator.java