@Override
    public int compareTo(BinlogOffset that) {
        // the NON_STOPPING is the max offset
        if (that.getOffsetKind() == NON_STOPPING && this.getOffsetKind() == NON_STOPPING) {
            return 0;
        }
        if (this.getOffsetKind() == NON_STOPPING) {
            return 1;
        }
        if (that.getOffsetKind() == NON_STOPPING) {
            return -1;
        }

        String gtidSetStr = this.getGtidSet();
        String targetGtidSetStr = that.getGtidSet();
        if (StringUtils.isNotEmpty(targetGtidSetStr)) {
            // The target offset uses GTIDs, so we ideally compare using GTIDs ...
            if (StringUtils.isNotEmpty(gtidSetStr)) {
                // Both have GTIDs, so base the comparison entirely on the GTID sets.
                GtidSet gtidSet = new GtidSet(gtidSetStr);
                GtidSet targetGtidSet = new GtidSet(targetGtidSetStr);
                if (gtidSet.equals(targetGtidSet)) {
                    long restartSkipEvents = this.getRestartSkipEvents();
                    long targetRestartSkipEvents = that.getRestartSkipEvents();
                    return Long.compare(restartSkipEvents, targetRestartSkipEvents);
                }
                // The GTIDs are not an exact match, so figure out if this is a subset of the target
                // offset
                // ...
                return gtidSet.isContainedWithin(targetGtidSet) ? -1 : 1;
            }
            // The target offset did use GTIDs while this did not use GTIDs. So, we assume
            // that this offset is older since GTIDs are often enabled but rarely disabled.
            // And if they are disabled,
            // it is likely that this offset would not include GTIDs as we would be trying
            // to read the binlog of a
            // server that no longer has GTIDs. And if they are enabled, disabled, and re-enabled,
            // per
            // https://dev.mysql.com/doc/refman/5.7/en/replication-gtids-failover.html all properly
            // configured slaves that
            // use GTIDs should always have the complete set of GTIDs copied from the master, in
            // which case
            // again we know that this offset not having GTIDs is before the target offset ...
            return -1;
        } else if (StringUtils.isNotEmpty(gtidSetStr)) {
            // This offset has a GTID but the target offset does not, so per the previous paragraph
            // we
            // assume that previous
            // is not at or before ...
            return 1;
        }

        // Both offsets are missing GTIDs. Look at the servers ...
        long serverId = this.getServerId();
        long targetServerId = that.getServerId();

        if (serverId != targetServerId) {
            // These are from different servers, and their binlog coordinates are not related. So
            // the only thing we can do
            // is compare timestamps, and we have to assume that the server timestamps can be
            // compared ...
            long timestamp = this.getTimestampSec();
            long targetTimestamp = that.getTimestampSec();
            return Long.compare(timestamp, targetTimestamp);
        }

        // First compare the MySQL binlog filenames
        if (this.getFilename().compareToIgnoreCase(that.getFilename()) != 0) {
            return this.getFilename().compareToIgnoreCase(that.getFilename());
        }

        // The filenames are the same, so compare the positions
        if (this.getPosition() != that.getPosition()) {
            return Long.compare(this.getPosition(), that.getPosition());
        }

        // The positions are the same, so compare the completed events in the transaction ...
        if (this.getRestartSkipEvents() != that.getRestartSkipEvents()) {
            return Long.compare(this.getRestartSkipEvents(), that.getRestartSkipEvents());
        }

        // The completed events are the same, so compare the row number ...
        return Long.compare(this.getRestartSkipRows(), that.getRestartSkipRows());
    }
This method is inspired by {@link io.debezium.relational.history.HistoryRecordComparator}.
inlong-sort/sort-flink/sort-flink-v1.13/sort-connectors/mysql-cdc/src/main/java/org/apache/inlong/sort/cdc/mysql/source/offset/BinlogOffset.java