docs: @Override
      public synchronized long commit(long offset) {
          if (offset <= submittedOffset && offset > safeOffset) {
              int ggap = (int)Math.min(committedOffsets.length, offset-safeOffset);
              if (ggap == committedOffsets.length) {
                  LOGGER.rateLimitedLog(LOG_SUPPRESSION_INTERVAL_SECONDS ,Level.WARN,
                          null, "CommitTracker moving topic commit point from %d to %d for topic "
                                  + topic + " partition " + partition + " group:" + consumerGroup, safeOffset, (offset - committedOffsets.length + 1)
                          );
                  safeOffset = offset - committedOffsets.length + 1;
                  committedOffsets[idx(safeOffset)] = safeOffset;
              }
              committedOffsets[idx(offset)] = offset;
              while (ggap > 0 && committedOffsets[idx(safeOffset)]+1 == committedOffsets[idx(safeOffset+1)]) {
                  ++safeOffset;
              }
              if (offerOffset >=0 && (offerOffset-safeOffset) < committedOffsets.length) {
                  offerOffset = -1L;
                  notify();
              }
          }
  
          if (offset == firstOffset) {
              firstOffsetCommitted = true;
          }
          return safeOffset;
      }