def METHOD_NAME(task_instance, execution_date, **kwargs):
    sentry_sdk.init()
    extract_path = task_instance.xcom_pull(task_ids="california_transit_gtfs_datasets")
    print(f"loading raw airtable records from {extract_path}")
    fs = get_fs()
    file = GCSFileInfo(**fs.info(extract_path))
    with fs.open(extract_path, "rb") as f:
        content = gzip.decompress(f.read())
    # I don't love this; this should be handled via metadata
    extract = AirtableGTFSDataExtract(
        filename=file.filename,
        ts=pendulum.parse(file.partition["ts"], exact=True),
        records=[
            AirtableGTFSDataRecord(**json.loads(row))
            for row in content.decode().splitlines()
        ],
    )
    valid, invalid, skipped = gtfs_datasets_to_extract_configs(extract)
    msg = f"{len(extract.records)=} {len(valid)=} {len(skipped)=} {len(invalid)=}"
    print(msg)
    print("Invalid records:")
    for record, error in invalid:
        print(record.name, record.pipeline_url, error)
    print("Skipped records:")
    for record in skipped:
        print(record.name, record.pipeline_url)
    # TODO: we should probably be configuring these alerts via Grafana but we need Pushgateway for that
    invalid_threshold_pct = 0.05
    if len(invalid) / len(extract.records) > invalid_threshold_pct:
        raise RuntimeError(f"more than {invalid_threshold_pct}% invalid records")
    # should update this if we end up with lots of historical records
    # as of 2023-03-29 we're at 5%
    skipped_threshold_pct = 0.15
    if len(skipped) / len(extract.records) > skipped_threshold_pct:
        raise RuntimeError(f"more than {skipped_threshold_pct}% skipped records")
    jsonl_content = gzip.compress("\n".join(record.json() for record in valid).encode())
    GTFSDownloadConfigExtract(
        filename="configs.jsonl.gz",
        ts=extract.ts,
        records=valid,
    ).save_content(content=jsonl_content, fs=fs)