dev/merge_spark_pr.py - master - githubmirror/spark

dev/merge_spark_pr.py
1 946 строк79 KB
Uros Bojanic
[SPARK-58638][INFRA] Recognize Examples as a primary component tag in the merge script
07 авг 2026, 20:02
07 авг 2026, 20:02f572940
#!/usr/bin/env python3

#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

# Utility for creating well-formed pull request merges and pushing them to Apache
# Spark.
#   usage: ./merge_spark_pr.py    (see config env vars below)
#
# Rolling integration branches branch-M.x (e.g. branch-4.x; later branch-5.x after 5.0.0)
# absorb work for the next minors on major M. Maintenance branches branch-M.N are cut from
# branch-M.x. Cherry-pick prompts sort by semver via semver_branch_rank (branch-M.x before branch-M.N).
#
# Fix Version defaults when resolving JIRA aim for the minimal set of versions that tells users
# which Spark releases will contain the commit, leveraging the Upstream-First backporting policy
# (cherry-picks flow master -> branch-M.x -> branch-M.N). master contributes the greatest
# unreleased N.0.0, but is omitted when any branch-M.x is in the merge set (a cherry-pick to
# branch-M.x has already landed on master). Each branch-M.x contributes that major's greatest
# unreleased minor.0, but is omitted when a sibling branch-M.N contributes M.N.0 (M.N.0 still
# unreleased means branch-M.N is still a subset of branch-M.x). Patch merges on branch-M.N use
# unreleased versions starting with M.N.
# This utility assumes you already have a local Spark git folder and that you
# have added remotes corresponding to both (i) the github apache Spark
# mirror and (ii) the apache git repo.

import json
import os
import re
import subprocess
import sys
import traceback
from typing import List
from urllib.request import urlopen
from urllib.request import Request
from urllib.error import HTTPError

# Shared with dev/pr_merge_status.py so the two committer tools agree on where a PR landed.
# Importable because Python puts this script's own directory first on sys.path.
from spark_merge_footer import branches_with_merge_footer as _branches_with_merge_footer
from spark_merge_footer import has_merge_footer

try:
    import jira.client

    JIRA_IMPORTED = True
except ImportError:
    JIRA_IMPORTED = False

# Location of your Spark git development area
SPARK_HOME = os.environ.get("SPARK_HOME", os.getcwd())
# Remote name which points to the Github site
PR_REMOTE_NAME = os.environ.get("PR_REMOTE_NAME", "apache-github")
# Remote name which points to Apache git
PUSH_REMOTE_NAME = os.environ.get("PUSH_REMOTE_NAME", "apache")
# ASF JIRA username
JIRA_USERNAME = os.environ.get("JIRA_USERNAME", "")
# ASF JIRA password
JIRA_PASSWORD = os.environ.get("JIRA_PASSWORD", "")
# ASF JIRA access token
# If it is configured, username and password are dismissed
# Go to https://issues.apache.org/jira/secure/ViewProfile.jspa -> Personal Access Tokens for
# your own token management.
JIRA_ACCESS_TOKEN = os.environ.get("JIRA_ACCESS_TOKEN")
# Connection timeout (in seconds) for the JIRA client. Raise this if the default is too
# short for your network, e.g. when the TLS handshake goes through a slow proxy.
JIRA_CONNECT_TIMEOUT = float(os.environ.get("JIRA_CONNECT_TIMEOUT", "3.05"))
# OAuth key used for issuing requests against the GitHub API. If this is not defined, then requests
# will be unauthenticated. You should only need to configure this if you find yourself regularly
# exceeding your IP's unauthenticated request rate limit. You can create an OAuth key at
# https://github.com/settings/tokens. This script only requires the "public_repo" scope.
GITHUB_OAUTH_KEY = os.environ.get("GITHUB_OAUTH_KEY")


GITHUB_BASE = "https://github.com/apache/spark/pull"
GITHUB_COMMIT_BASE = "https://github.com/apache/spark/commit"
GITHUB_API_BASE = "https://api.github.com/repos/apache/spark"
JIRA_BASE = "https://issues.apache.org/jira/browse"
JIRA_API_BASE = "https://issues.apache.org/jira"
# Prefix added to temporary branches
BRANCH_PREFIX = "PR_TOOL"
# Set to a truthy value to skip the check that compares this script against apache/spark master.
# Intended for committers iterating on the merge script itself, where a local diff is expected.
SKIP_VERSION_CHECK = os.environ.get("SKIP_VERSION_CHECK", "")
# Path of this script relative to the repo root, used to fetch the canonical copy from master.
MERGE_SCRIPT_REPO_PATH = "dev/merge_spark_pr.py"


def semver_branch_rank(name):
    """
    Sort key for maintenance branches: higher merges / cherry-picks first.

    Goals:
      - Numeric Spark branches branch-M.N sort by (M, minor), not lexicographically
        (branch-4.11 is newer than branch-4.2).
      - Rolling integration branches branch-M.x rank before branch-M.N for the same M so they are
        suggested first among targets on major M.

    >>> semver_branch_rank("branch-5.x") > semver_branch_rank("branch-5.0") > semver_branch_rank("branch-4.11")
    True
    >>> semver_branch_rank("branch-5.x") > semver_branch_rank("branch-5.2")
    True
    >>> semver_branch_rank("branch-4.x") > semver_branch_rank("branch-4.2")
    True
    >>> semver_branch_rank("branch-4.2") > semver_branch_rank("branch-4.11")
    False
    """
    integration = re.match(r"^branch-(\d+)\.x$", name)
    if integration:
        return int(integration.group(1)), float("inf")
    matched = re.match(r"^branch-(\d+)\.(\d+)$", name)
    if matched:
        return int(matched.group(1)), int(matched.group(2))
    return (-1, -1)


def _semver_max_version(names):
    """
    Highest dotted version by numeric semver (SPARK Fix Version naming).

    >>> _semver_max_version(["4.2.1", "4.3.0"])
    '4.3.0'
    >>> _semver_max_version(["5.0.0", "6.0.0"])
    '6.0.0'
    """
    if not names:
        return None
    parsed = [(tuple(int(p) for p in n.split(".")), n) for n in names]
    return max(parsed)[1]


def _integration_major_from_branch(branch_name):
    """If branch-name is branch-M.x, return integer M; else None."""
    m = re.match(r"^branch-(\d+)\.x$", branch_name)
    return int(m.group(1)) if m else None


def _first_integration_major(merge_branches):
    """First branch-M.x on merge_branches in list order (deterministic).

    >>> _first_integration_major(["master", "branch-5.x", "branch-4.x"])
    5
    >>> _first_integration_major(["master", "branch-4.2"]) is None
    True
    """
    for ref in merge_branches:
        mj = _integration_major_from_branch(ref)
        if mj is not None:
            return mj
    return None


def _sort_version_names_lex_desc(names):
    """Same ordering as resolve_jira_issue (reverse lexicographic on Fix Version name)."""
    return sorted(names, reverse=True)


def compute_merge_default_fix_versions(merge_branches, unreleased_version_names):
    """
    Build suggested SPARK Fix Version names from merged git refs (no JIRA I/O).

    unreleased_version_names: x.y.z strings marked unreleased in JIRA.

    Returns (default_version_names, warning_messages).

    The output is the minimal set of Fix Versions that tells users which Spark releases will
    contain the commit, leveraging the Upstream-First backporting policy (cherry-picks flow
    master -> branch-M.x -> branch-M.N):
      - master contributes the greatest unreleased N.0.0;
      - branch-M.x with master contributes that major's greatest unreleased minor.0;
      - branch-M.N contributes its greatest unreleased M.N.patch.
    Redundant entries are then suppressed: master's N.0.0 is dropped when any branch-M.x is in
    the merge set (a cherry-pick to branch-M.x has already landed on master); branch-M.x's
    minor.0 is dropped when a branch-M.N for the same major contributes M.N.0 (M.N.0 still
    unreleased means branch-M.N is still a subset of branch-M.x).

    >>> compute_merge_default_fix_versions(["master"], ["4.3.0", "5.0.0", "6.0.0"])[0]
    ['6.0.0']

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-4.x"], ["6.0.0", "5.0.0", "4.3.0", "4.2.1"])[0]
    ['4.3.0']

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-5.x"],
    ...     ["7.0.0", "6.0.0", "5.2.0", "5.1.2", "5.1.1", "5.0.1"],
    ... )[0]
    ['5.2.0']

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-3.5"],
    ...     ["6.0.0", "4.1.0", "3.5.2", "3.5.1", "3.5.0"],
    ... )[0]
    ['6.0.0', '3.5.2']

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-1.1", "branch-1.0"],
    ...     ["2.0.0", "1.1.1", "1.1.0", "1.0.1", "1.0.0"],
    ... )[0]
    ['2.0.0', '1.1.1', '1.0.1']

    >>> d, w = compute_merge_default_fix_versions(["master"], ["4.3.0", "4.2.1"])
    >>> (d, len(w), "N.0.0" in w[0])
    ([], 1, True)

    >>> compute_merge_default_fix_versions(["branch-4.x"], ["4.3.0"])
    ([], [])

    >>> d, w = compute_merge_default_fix_versions(["branch-4.99"], ["4.3.0"])
    >>> d == [] and len(w) == 1 and "branch-4.99" in w[0]
    True

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-5.x", "branch-4.x"], ["5.2.0", "4.3.0", "6.0.0"]
    ... )[0]
    ['5.2.0', '4.3.0']

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-4.x", "branch-4.x"], ["6.0.0", "4.3.0"]
    ... )[0]
    ['4.3.0']

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-4.x", "branch-4.2"], ["5.0.0", "4.3.0", "4.2.0"]
    ... )[0]
    ['4.2.0']

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-4.x", "branch-4.2"], ["5.0.0", "4.3.0", "4.2.1"]
    ... )[0]
    ['4.3.0', '4.2.1']

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-4.2"], ["5.0.0", "4.2.0"]
    ... )[0]
    ['5.0.0', '4.2.0']

    >>> compute_merge_default_fix_versions(
    ...     ["master", "branch-4.2"], ["5.0.0", "4.2.1"]
    ... )[0]
    ['5.0.0', '4.2.1']
    """
    names = _sort_version_names_lex_desc(list(unreleased_version_names))
    has_integration_branch = any(
        _integration_major_from_branch(b) is not None for b in merge_branches
    )

    # Collect each merge branch's candidate Fix Version. We track (branch, version) so we can
    # later suppress redundant entries (master, branch-M.x) based on which branches contributed.
    contributions = []
    warnings = []

    for b in merge_branches:
        if b == "master":
            majors = [n for n in names if re.match(r"^\d+\.0\.0$", n)]
            chosen = _semver_max_version(majors)
            if chosen:
                contributions.append((b, chosen))
            else:
                warnings.append(
                    "No unreleased N.0.0 Fix Version found in JIRA for master; "
                    "enter comma-separated Fix Version(s) manually when prompted."
                )
            continue
        line_major = _integration_major_from_branch(b)
        if line_major is not None:
            if "master" not in merge_branches:
                continue
            line_versions = [n for n in names if re.match(r"^%s\.\d+\.\d+$" % line_major, n)]
            chosen = _semver_max_version(line_versions)
            if chosen:
                contributions.append((b, chosen))
            else:
                warnings.append(
                    "Could not infer an unreleased Spark %s (minor.maintenance) fix version "
                    "for branch-%s.x + master merge; enter version(s) manually when prompted."
                    % (line_major, line_major)
                )
            continue
        prefix = b.replace("branch-", "")
        found_versions = [n for n in names if n.startswith(prefix)]
        chosen = _semver_max_version(found_versions)
        if chosen:
            contributions.append((b, chosen))
        else:
            warnings.append(
                "Target version for %s is not found on JIRA, it may be archived or "
                "not created. Skipping it." % b
            )

    # Majors where a release branch's M.N.0 contribution makes the sibling branch-M.x's
    # minor.0 redundant: an unreleased M.N.0 means branch-M.N has not diverged from branch-M.x
    # yet, so anything that's on branch-M.N is also on branch-M.x.
    suppress_integration_majors = set()
    for b, v in contributions:
        if b == "master" or _integration_major_from_branch(b) is not None:
            continue
        m = re.match(r"^(\d+)\.\d+\.0$", v)
        if m:
            suppress_integration_majors.add(int(m.group(1)))

    def keep(item):
        b, _ = item
        if b == "master":
            return not has_integration_branch
        major = _integration_major_from_branch(b)
        if major is not None:
            return major not in suppress_integration_majors
        return True

    filtered = [item for item in contributions if keep(item)]
    return list(dict.fromkeys(v for _, v in filtered)), warnings


def red(text):
    return "\033[91m%s\033[0m" % text


def bold(text):
    return "\033[1m%s\033[0m" % text


def print_error(msg):
    print(red(msg))


def bold_input(prompt) -> str:
    return input("\033[1m%s\033[0m" % prompt)


def get_input(prompt, options, bold=True, ignore_case=True):
    """
    Get input from the user until a valid answer is provided.

    Args:
        prompt: The prompt to display to the user.
        options:
            * A dictionary of "option: accepted answer" to choose from.
            * A list of options.
            * A regex pattern.
        bold: Whether to use bold formatting for the prompt.
        ignore_case: Whether to ignore case when comparing the answer to the options.

    Returns:
        If options is provided - the valid answer from the user.
        If regex is provided
            * If no group is specified - the whole matched string.
            * If one group is specified - the first group.
            * If multiple groups are specified - a tuple of the groups.
    """

    # Normalize options
    if isinstance(options, (list, tuple)):
        options = {option: [option] for option in options}

    if isinstance(options, dict):
        for option, acceptable_answer in options.items():
            if ignore_case:
                options[option] = [answer.lower() for answer in acceptable_answer]

    while True:
        if bold:
            answer = bold_input(prompt)
        else:
            answer = input(prompt)

        answer = answer.strip()
        if ignore_case:
            answer = answer.lower()

        if isinstance(options, str):
            if (m := re.match(options, answer)) is not None:
                groups = m.groups()
                if len(groups) == 0:
                    return m.group(0)
                if len(groups) == 1:
                    return groups[0]
                else:
                    return groups
        else:
            for option, acceptable_answer in options.items():
                if answer in acceptable_answer:
                    return option


def get_json(url):
    try:
        request = Request(url)
        if GITHUB_OAUTH_KEY:
            request.add_header("Authorization", "token %s" % GITHUB_OAUTH_KEY)
        return json.load(urlopen(request))
    except HTTPError as e:
        if "X-RateLimit-Remaining" in e.headers and e.headers["X-RateLimit-Remaining"] == "0":
            print_error(
                "Exceeded the GitHub API rate limit; see the instructions in "
                + "dev/merge_spark_pr.py to configure an OAuth token for making authenticated "
                + "GitHub requests."
            )
        elif e.code == 401:
            print_error(
                "GITHUB_OAUTH_KEY is invalid or expired. Please regenerate a new one with "
                + "at least the 'public_repo' scope on https://github.com/settings/tokens and "
                + "update your local settings before you try again."
            )
        else:
            print_error("Unable to fetch URL, exiting: %s" % url)
        sys.exit(-1)


def merge_commit_candidates(pr_events):
    """Split `pr_events` into (closed_commits, referenced_commits), each oldest-first.

    Ordered by time so that a PR reopened and merged again yields its latest merge last.

    >>> merge_commit_candidates([{"event": "closed", "commit_id": "a", "created_at": "t2"},
    ...                          {"event": "referenced", "commit_id": "b", "created_at": "t1"}])
    (['a'], ['b'])
    >>> merge_commit_candidates([{"event": "closed", "commit_id": None, "created_at": "t1"}])
    ([], [])
    >>> merge_commit_candidates([{"event": "referenced", "commit_id": "c", "created_at": "t2"},
    ...                          {"event": "referenced", "commit_id": "b", "created_at": "t1"}])
    ([], ['b', 'c'])
    """

    def commits_of(event_name):
        matched = [e for e in pr_events if e["event"] == event_name and e["commit_id"] is not None]
        return [e["commit_id"] for e in sorted(matched, key=lambda x: x["created_at"])]

    return commits_of("closed"), commits_of("referenced")


def find_merge_commit(pr_num, pr_events):
    """Return (hash, message) of the commit that merged `pr_num`, or (None, None).

    GitHub attributes the merge commit to the `closed` event only when that commit lands
    on the default branch (master), because the "Closes #N" keyword in the commit message
    is what closes the PR and the keyword is honored only there. A PR merged into any
    other branch -- e.g. one opened against a rolling branch-M.x -- is instead closed by
    this script through the API, and that `closed` event carries no commit, so the merge
    survives only as a `referenced` event. Prefer the `closed` commit, which GitHub itself
    linked; otherwise fall back to `referenced` events, which are also raised by any commit
    merely mentioning the PR, so confirm each against the merge footer `merge_pr` generates.
    """

    def message_of(commit_hash):
        return get_json("%s/commits/%s" % (GITHUB_API_BASE, commit_hash))["commit"]["message"]

    closed_commits, referenced_commits = merge_commit_candidates(pr_events)
    if closed_commits:
        return closed_commits[-1], message_of(closed_commits[-1])

    for commit_hash in reversed(referenced_commits):
        message = message_of(commit_hash)
        if has_merge_footer(message, pr_num):
            return commit_hash, message
    return None, None


def close_pr(pr_num):
    url = "%s/pulls/%s" % (GITHUB_API_BASE, pr_num)
    data = json.dumps({"state": "closed"}).encode("utf-8")
    request = Request(url, data=data, method="PATCH")
    request.add_header("Content-Type", "application/json")
    request.add_header("Accept", "application/vnd.github+json")
    if GITHUB_OAUTH_KEY:
        request.add_header("Authorization", "token %s" % GITHUB_OAUTH_KEY)
    try:
        return json.load(urlopen(request))
    except HTTPError as e:
        print_error("Failed to close PR #%s: HTTP %s %s" % (pr_num, e.code, e.reason))
        return None


def comment_pr(pr_num, body):
    url = "%s/issues/%s/comments" % (GITHUB_API_BASE, pr_num)
    data = json.dumps({"body": body}).encode("utf-8")
    request = Request(url, data=data, method="POST")
    request.add_header("Content-Type", "application/json")
    request.add_header("Accept", "application/vnd.github+json")
    if GITHUB_OAUTH_KEY:
        request.add_header("Authorization", "token %s" % GITHUB_OAUTH_KEY)
    try:
        return json.load(urlopen(request))
    except HTTPError as e:
        print_error("Failed to comment on PR #%s: HTTP %s %s" % (pr_num, e.code, e.reason))
        return None


def post_merge_comment(pr_num, merged_commits):
    """Post a comment on the PR recording every branch the change landed on and a
    link to the resulting commit, so the merge is traceable from the PR page.

    ``merged_commits`` is an ordered list of (branch, commit_hash) pairs, the merge
    sink first followed by each cherry-pick target.
    """
    if not merged_commits:
        return
    lines = [
        "- merged into %s %s/%s" % (ref, GITHUB_COMMIT_BASE, commit_hash)
        for ref, commit_hash in merged_commits
    ]
    summary = "**Merge Summary:**\n" + "\n".join(lines)
    attribution = "*Posted by `merge_spark_pr.py`*"
    body = "%s\n\n%s" % (summary, attribution)
    print(
        "\n%s\n\n%s\n%s"
        % (bold("Posting merge comment on PR #%s:" % pr_num), bold(summary), attribution)
    )
    if not GITHUB_OAUTH_KEY:
        print_error("GITHUB_OAUTH_KEY is not set; skipping the merge comment.")
        return
    comment_pr(pr_num, body)


def fail(msg):
    print_error(msg)
    clean_up()
    sys.exit(-1)


def run_cmd(cmd):
    print(cmd)
    if isinstance(cmd, list):
        return subprocess.check_output(cmd).decode("utf-8")
    else:
        return subprocess.check_output(cmd.split(" ")).decode("utf-8")


def continue_maybe(prompt, cherry=False):
    if get_input(f"{prompt} (y/N): ", ["y", "n", ""]) != "y":
        if cherry:
            try:
                run_cmd("git cherry-pick --abort")
            except Exception:
                print_error("Unable to abort and get back to the state before cherry-pick")
        fail("Okay, exiting")


def clean_up():
    if "original_head" in globals():
        print("Restoring head pointer to %s" % original_head)
        run_cmd("git checkout %s" % original_head)

        branches = run_cmd("git branch").replace(" ", "").split("\n")

        for branch in list(filter(lambda x: x.startswith(BRANCH_PREFIX), branches)):
            print("Deleting local branch %s" % branch)
            run_cmd("git branch -D %s" % branch)


# merge the requested PR and return the merge hash
def merge_pr(pr_num, target_ref, title, body, pr_repo_desc, pr_author, co_authors):
    pr_branch_name = "%s_MERGE_PR_%s" % (BRANCH_PREFIX, pr_num)
    target_branch_name = "%s_MERGE_PR_%s_%s" % (BRANCH_PREFIX, pr_num, target_ref.upper())
    run_cmd("git fetch %s pull/%s/head:%s" % (PR_REMOTE_NAME, pr_num, pr_branch_name))
    run_cmd("git fetch %s %s:%s" % (PUSH_REMOTE_NAME, target_ref, target_branch_name))
    run_cmd("git checkout %s" % target_branch_name)

    had_conflicts = False
    try:
        run_cmd(["git", "merge", pr_branch_name, "--squash"])
    except Exception as e:
        msg = "Error merging: %s\nWould you like to manually fix-up this merge?" % e
        continue_maybe(msg)
        msg = "Okay, please fix any conflicts and 'git add' conflicting files... Finished?"
        continue_maybe(msg)
        had_conflicts = True

    # Use PR author as the primary author for proper GitHub attribution
    primary_author = bold_input(
        'Enter primary author in the format of "name <email>" [%s]: ' % pr_author
    )
    if primary_author == "":
        primary_author = pr_author

    merge_message_flags = []

    merge_message_flags += ["-m", title]
    if body is not None:
        # We remove @ symbols from the body to avoid triggering e-mails
        # to people every time someone creates a public fork of Spark.
        merge_message_flags += ["-m", body.replace("@", "")]

    committer_name = run_cmd("git config --get user.name").strip()
    committer_email = run_cmd("git config --get user.email").strip()

    if had_conflicts:
        message = "This patch had conflicts when merged, resolved by\nCommitter: %s <%s>" % (
            committer_name,
            committer_email,
        )
        merge_message_flags += ["-m", message]

    # The string "Closes #%s" string is required for GitHub to correctly close the PR
    merge_message_flags += ["-m", "Closes #%s from %s." % (pr_num, pr_repo_desc)]

    authors = "Authored-by:" if len(co_authors) == 0 else "Lead-authored-by:"
    authors += " %s" % primary_author
    if len(co_authors) > 0:
        authors += "\n" + "\n".join(["Co-authored-by: %s" % a for a in co_authors])
    authors += "\n" + "Signed-off-by: %s <%s>" % (committer_name, committer_email)

    merge_message_flags += ["-m", authors]

    run_cmd(["git", "commit", '--author="%s"' % primary_author] + merge_message_flags)

    continue_maybe(
        "Merge complete (local ref %s). Push to %s?" % (target_branch_name, PUSH_REMOTE_NAME)
    )

    try:
        run_cmd("git push %s %s:%s" % (PUSH_REMOTE_NAME, target_branch_name, target_ref))
    except Exception as e:
        clean_up()
        print_error("Exception while pushing: %s" % e)

    merge_hash = run_cmd("git rev-parse %s" % target_branch_name).strip()
    clean_up()
    print("Pull request #%s merged!" % pr_num)
    print("Merge hash: %s" % merge_hash)
    return merge_hash


def _do_cherry_pick(pr_num, merge_hash, pick_ref):
    """Cherry-pick `merge_hash` onto `pick_ref` and push.

    Returns the (pushed ref, pushed commit hash) pair.
    """
    pick_branch_name = "%s_PICK_PR_%s_%s" % (BRANCH_PREFIX, pr_num, pick_ref.upper())

    run_cmd("git fetch %s %s:%s" % (PUSH_REMOTE_NAME, pick_ref, pick_branch_name))
    run_cmd("git checkout %s" % pick_branch_name)

    try:
        run_cmd("git cherry-pick -sx %s" % merge_hash)
    except Exception as e:
        msg = "Error cherry-picking: %s\nWould you like to manually fix-up this merge?" % e
        continue_maybe(msg, True)
        msg = "Okay, please fix any conflicts and finish the cherry-pick. Finished?"
        continue_maybe(msg, True)

    continue_maybe(
        "Pick complete (local ref %s). Push to %s?" % (pick_branch_name, PUSH_REMOTE_NAME)
    )

    try:
        run_cmd("git push %s %s:%s" % (PUSH_REMOTE_NAME, pick_branch_name, pick_ref))
    except Exception as e:
        fail("Exception while pushing: %s" % e)

    pick_hash = run_cmd("git rev-parse %s" % pick_branch_name).strip()
    clean_up()

    print("Pull request #%s picked into %s!" % (pr_num, pick_ref))
    print("Pick hash: %s" % pick_hash)
    return pick_ref, pick_hash


def branches_with_merge_footer(pr_num, branch_names):
    """Release branches from `branch_names` that already carry `pr_num`'s merge footer.

    Thin wrapper over the shared reader in `spark_merge_footer`, adding this script's own
    policy: a git failure here must not abort a merge that may already have pushed, so it
    warns and reports nothing rather than exiting. Per that module's refresh policy no fetch
    is issued, so a backport not yet fetched into PUSH_REMOTE_NAME's tracking refs is simply
    not reported -- the committer is still prompted and can type any branch.
    """
    try:
        landed = _branches_with_merge_footer(
            pr_num, PUSH_REMOTE_NAME, lambda args: run_cmd(["git"] + args)
        )
    except Exception as e:
        print_error("Could not scan for existing backports of #%s (%s)." % (pr_num, e))
        return []
    # Keep branch_names' newest-first order, and drop anything not a known release branch.
    return [b for b in branch_names if b in landed]


def default_pick_branch(branch_names, already_picked):
    """Highest-ranked release branch that has not already received the change, or None.

    `branch_names` is ordered newest-first (see `semver_branch_rank`) and `already_picked`
    holds the branches the change is known to be on, so the prompt never defaults to a
    branch where the cherry-pick would come up empty. Returns None when every known branch
    already has it, so callers can say so instead of offering an empty pick.

    >>> default_pick_branch(["branch-4.x", "branch-4.3", "branch-4.2"], ("branch-4.x",))
    'branch-4.3'
    >>> default_pick_branch(["branch-4.x", "branch-4.3"], ())
    'branch-4.x'
    >>> default_pick_branch(["branch-4.x"], ("branch-4.x",)) is None
    True
    """
    remaining = [b for b in branch_names if b not in already_picked]
    return remaining[0] if remaining else None


def _upstream_first_sibling(target_ref, pick_ref, branch_names, already_picked):
    """Return the sibling branch-M.x if Upstream-First should prompt, else None.

    The policy only applies when the PR was merged into master: that's the only case
    where the committer can type branch-M.N at the cherry-pick prompt and bypass the
    rolling branch-M.x. When the PR was opened against branch-M.x the merge itself
    lands there (nothing to bypass), and when it was opened against branch-M.N the
    author already chose per-branch scope.

    >>> _upstream_first_sibling("master", "branch-4.2", ["branch-4.x", "branch-4.2"], ())
    'branch-4.x'
    >>> _upstream_first_sibling("master", "branch-4.2", ["branch-4.x", "branch-4.2"],
    ...                         ("branch-4.x",))
    >>> _upstream_first_sibling("master", "branch-4.x", ["branch-4.x"], ())
    >>> _upstream_first_sibling("master", "branch-4.99", ["branch-4.2"], ())
    >>> _upstream_first_sibling("branch-4.x", "branch-4.2", ["branch-4.x", "branch-4.2"], ())
    >>> _upstream_first_sibling("branch-4.2", "branch-3.5", ["branch-4.x", "branch-3.5"], ())
    """
    if target_ref != "master":
        return None
    m = re.match(r"^branch-(\d+)\.(\d+)$", pick_ref)
    if not m:
        return None
    candidate = "branch-%s.x" % m.group(1)
    if candidate in branch_names and candidate not in already_picked:
        return candidate
    return None


def cherry_pick(pr_num, merge_hash, default_branch, branch_names, target_ref, already_picked=()):
    """Prompt for a target branch and cherry-pick `merge_hash` onto it.

    Enforces the Upstream-First policy (see header comment) via
    `_upstream_first_sibling`: when the PR was merged into master and the committer
    types a branch-M.N target while branch-M.x is also a known release branch AND
    has not already received this commit, prompt to confirm whether to pick into
    BOTH (the policy-compliant default) or branch-M.N only (treated as a
    maintenance-only bugfix). Returns the list of (ref, commit_hash) pairs actually
    picked into, so the main loop can advance its remaining-branches list correctly
    and record each backport commit for the merge comment.
    """
    while True:
        pick_ref = bold_input(f"Enter a branch name [{default_branch}]: ")
        if pick_ref == "":
            pick_ref = default_branch
        if pick_ref in branch_names:
            break
        valid_branches = ", ".join(branch_names)
        print_error(
            f"'{pick_ref}' is not a known release branch. "
            f"Valid branches: {valid_branches}. Please try again."
        )

    sibling_x = _upstream_first_sibling(target_ref, pick_ref, branch_names, already_picked)
    if sibling_x is not None:
        print()
        print("=" * 80)
        print(
            "Upstream-First policy: non-bugfix commits on %s should also land on %s."
            % (pick_ref, sibling_x)
        )
        print(
            "If this is a %s-only maintenance bugfix, you may pick %s alone." % (pick_ref, pick_ref)
        )
        print("Otherwise, pick both (%s first, then %s)." % (sibling_x, pick_ref))
        print("=" * 80)
        choice = get_input(
            f"Pick into [b]oth {sibling_x} + {pick_ref} / [o]nly {pick_ref} / [a]bort (default: both): ",
            {"b": ["b", "both", ""], "o": ["o", "only"], "a": ["a", "abort"]},
        )
        if choice == "b":
            picked_x = _do_cherry_pick(pr_num, merge_hash, sibling_x)
            picked_n = _do_cherry_pick(pr_num, merge_hash, pick_ref)
            return [picked_x, picked_n]
        elif choice == "o":
            return [_do_cherry_pick(pr_num, merge_hash, pick_ref)]
        elif choice == "a":
            fail("Aborted by user at Upstream-First policy prompt.")
        else:
            fail("Unrecognized choice %r; aborting." % choice)

    return [_do_cherry_pick(pr_num, merge_hash, pick_ref)]


# Common words carry no signal when comparing a PR title to a JIRA summary, so they are
# dropped before scoring. Kept deliberately small: over-aggressive stopword removal makes
# unrelated titles look similar. Component tags ([SQL], [CORE], ...) are stripped separately.
_SIMILARITY_STOPWORDS = frozenset(
    {
        "a",
        "an",
        "the",
        "to",
        "for",
        "of",
        "in",
        "on",
        "and",
        "or",
        "with",
        "is",
        "are",
        "be",
        "when",
        "should",
        "make",
        "add",
        "fix",
        "fixes",
        "support",
        "enable",
        "use",
        "using",
    }
)
_SIMILARITY_WORD_RE = re.compile(r"[a-z0-9]+")
# Below this Jaccard word-overlap score, warn that the PR title and JIRA summary look
# unrelated. Tuned to flag clear mismatches (an unrelated ticket scores ~0) while leaving
# paraphrases and reworded summaries (which still share key nouns) above the line.
SIMILARITY_WARN_THRESHOLD = 0.2


def title_similarity(pr_title, summary):
    """Jaccard word-overlap between a PR title and a JIRA summary, in [0.0, 1.0].

    Component/version tags ([SQL], [4.x], ...) and the leading SPARK id are stripped,
    words are lowercased, and common stopwords are dropped, so the score reflects the
    substantive words the two share. 1.0 means identical word sets; 0.0 means none in
    common (or an empty side).

    >>> title_similarity("[SPARK-1][SQL] Compute stable checksum", "Compute a stable checksum")
    1.0
    >>> title_similarity("[SPARK-1][SQL] Compute stable checksum", "Refactor the logging backend")
    0.0
    >>> round(title_similarity("[SPARK-1] Ceil and floor overflow", "Handle floor overflow"), 2)
    0.5

    A follow-up title usually describes the fix it adds, not the original ticket, so
    it scores low against the JIRA summary even though it references the right ticket
    (the caller skips the low-similarity warning for [FOLLOWUP] PRs -- see
    format_jira_verification):

    >>> title_similarity("[SPARK-1][FOLLOWUP] Fix a typo", "Add a cache")
    0.0

    A Revert PR keeps its title verbatim (the "Revert" prefix and quotes are not
    tags), so the extra word lowers the score but a genuine match still scores high:

    >>> round(title_similarity('Revert "[SPARK-1] Add a cache"', "Add a cache"), 2)
    0.5
    """

    def tokens(text):
        text = re.sub(r"\[[^\]]*\]", " ", text)
        words = _SIMILARITY_WORD_RE.findall(text.lower())
        return {w for w in words if w not in _SIMILARITY_STOPWORDS}

    a = tokens(pr_title)
    b = tokens(summary)
    if not a or not b:
        return 0.0
    return len(a & b) / len(a | b)


def format_jira_verification(
    pr_num, pr_title, jira_id, summary, status, issuetype, use_color=False, is_followup=False
):
    """Render the JIRA-vs-PR match block shown before merging.

    Places the PR title next to the linked ticket's summary so the committer can
    eyeball whether they match, flags a ticket that is already Resolved/Closed
    (a fresh merge should target an open ticket), and scores how much the PR title
    and the JIRA summary overlap -- a low score suggests the wrong ticket. Pure
    formatting so it is covered by the inline doctests. ``use_color`` wraps the
    warnings in the same red as ``print_error``; it is left off in the doctests so
    the expected output stays plain text.

    ``is_followup`` suppresses the low-similarity warning: a [FOLLOWUP] PR title
    describes the fix it adds, not the original ticket, so it legitimately scores
    low against the JIRA summary. The score is still shown for reference, but with
    a note instead of a warning so the expected divergence isn't flagged as a
    likely wrong ticket. The Resolved/Closed warning still applies.

    >>> print(format_jira_verification(
    ...     42,
    ...     "[SPARK-2222][SQL] Compute stable checksum",
    ...     "SPARK-1111",
    ...     "Refactor the logging backend",
    ...     "Resolved",
    ...     "Improvement",
    ... ))
    === Verify JIRA matches PR #42 ===
    PR title:  [SPARK-2222][SQL] Compute stable checksum
    JIRA SPARK-1111: Refactor the logging backend
      Status:  Resolved   <-- WARNING: already Resolved/Closed
      Type:    Improvement
      Match:   0.00   <-- WARNING: low title similarity, wrong ticket?

    >>> print(format_jira_verification(
    ...     42,
    ...     "[SPARK-2222][SQL] Compute stable checksum",
    ...     "SPARK-2222",
    ...     "Compute a stable checksum",
    ...     "In Progress",
    ...     "Bug",
    ... ))
    === Verify JIRA matches PR #42 ===
    PR title:  [SPARK-2222][SQL] Compute stable checksum
    JIRA SPARK-2222: Compute a stable checksum
      Status:  In Progress
      Type:    Bug
      Match:   1.00

    A [FOLLOWUP] title scores low but is not warned about (only noted):

    >>> print(format_jira_verification(
    ...     42,
    ...     "[SPARK-1][FOLLOWUP] Fix a typo",
    ...     "SPARK-1",
    ...     "Add a cache",
    ...     "In Progress",
    ...     "Bug",
    ...     is_followup=True,
    ... ))
    === Verify JIRA matches PR #42 ===
    PR title:  [SPARK-1][FOLLOWUP] Fix a typo
    JIRA SPARK-1: Add a cache
      Status:  In Progress
      Type:    Bug
      Match:   0.00   (FOLLOWUP: title intentionally differs, not checked)
    """
    status_warning = ""
    if status in ("Resolved", "Closed"):
        status_warning = "   <-- WARNING: already Resolved/Closed"
        if use_color:
            status_warning = red(status_warning)
    score = title_similarity(pr_title, summary)
    if is_followup:
        # A follow-up title describes its own fix, not the original ticket, so a low
        # score is expected and not a wrong-ticket signal. Note it, don't warn.
        match_suffix = "   (FOLLOWUP: title intentionally differs, not checked)"
    elif score < SIMILARITY_WARN_THRESHOLD:
        match_suffix = "   <-- WARNING: low title similarity, wrong ticket?"
        if use_color:
            match_suffix = red(match_suffix)
    else:
        match_suffix = ""
    return "\n".join(
        [
            "=== Verify JIRA matches PR #%s ===" % pr_num,
            "PR title:  %s" % pr_title,
            "JIRA %s: %s" % (jira_id, summary),
            "  Status:  %s%s" % (status, status_warning),
            "  Type:    %s" % issuetype,
            "  Match:   %.2f%s" % (score, match_suffix),
        ]
    )


def print_jira_issue_summary(issue):
    summary = "Summary\t\t%s\n" % issue.fields.summary
    assignee = issue.fields.assignee
    if assignee is not None:
        assignee = assignee.displayName
    assignee = "Assignee\t%s\n" % assignee
    status = "Status\t\t%s\n" % issue.fields.status.name
    components = "Components\t%s\n" % [x.name for x in issue.fields.components]
    url = "Url\t\t%s/%s\n" % (JIRA_BASE, issue.key)
    target_versions = "Affected\t%s\n" % [x.name for x in issue.fields.versions]
    fix_versions = ""
    if len(issue.fields.fixVersions) > 0:
        fix_versions = "Fixed\t\t%s\n" % [x.name for x in issue.fields.fixVersions]
    print("=== JIRA %s ===" % issue.key)
    print(
        "%s%s%s%s%s%s%s"
        % (summary, assignee, status, components, url, target_versions, fix_versions)
    )


def jira_components_from_title_tags(tags):
    """Canonical SPARK JIRA component names implied by PR-title component tags.

    Each tag is resolved through the component registry; a tag that maps to a
    JIRA component contributes that component's canonical name, whether primary
    or not (e.g. [SQL] -> "SQL", [TEST] -> "Tests"). Tags that are not JIRA
    components (status markers like [FOLLOWUP]/[MINOR], version tags like [4.X],
    or unknown tags) contribute nothing. Aliases normalize to the canonical
    name. The result preserves input order and is de-duplicated.

    >>> jira_components_from_title_tags(["SQL", "CORE"])
    ['SQL', 'Spark Core']
    >>> jira_components_from_title_tags(["PYTHON", "DOCS"])
    ['PySpark', 'Documentation']
    >>> jira_components_from_title_tags(["SQL", "TEST"])
    ['SQL', 'Tests']
    >>> jira_components_from_title_tags(["SQL", "FOLLOWUP", "4.X", "BOGUS"])
    ['SQL']
    >>> jira_components_from_title_tags(["SQL", "SQL"])
    ['SQL']
    """
    names = []
    for tag in tags:
        c = Component.find(tag)
        if c is not None and c.jira_name:
            names.append(c.jira_name)
    return list(dict.fromkeys(names))


def reconcile_jira_components(issue, title_components):
    """Prompt to sync JIRA components when they differ from the PR title.

    ``title_components`` is the list of normalized PR-title component tags (e.g.
    ["SQL", "TEST"]). Every tag that maps to a JIRA component -- primary or not,
    e.g. [SQL] -> "SQL" and [TEST] -> "Tests" -- is reconciled; tags with no JIRA
    component ([MINOR], [FOLLOWUP], version tags, unknown tags) are dropped. The
    mapped names are compared, as a set, against the issue's current components.
    On a mismatch, offer to overwrite JIRA with the PR title's components, append
    them to the existing ones, or keep JIRA unchanged (the default).
    """
    title_jira_components = jira_components_from_title_tags(title_components)
    if not title_jira_components:
        return
    current = [c.name for c in issue.fields.components]
    if set(current) == set(title_jira_components):
        return

    print()
    print("=" * 80)
    print("PR title components differ from JIRA %s:" % issue.key)
    print("  PR title: %s" % ", ".join(title_jira_components))
    print("  JIRA:     %s" % (", ".join(current) if current else "(none)"))
    print("=" * 80)
    choice = get_input(
        "[o]verwrite JIRA with PR title / [a]ppend PR title to JIRA / [k]eep JIRA as is "
        "(default: keep): ",
        {"o": ["o", "overwrite"], "a": ["a", "append"], "k": ["k", "keep", ""]},
    )
    if choice == "k":
        print("Keeping JIRA %s components unchanged." % issue.key)
        return
    if choice == "o":
        new_names = list(title_jira_components)
    elif choice == "a":
        # Append the PR title's components, keeping the existing ones first.
        new_names = list(dict.fromkeys(current + title_jira_components))

    try:
        issue.update(fields={"components": [{"name": n} for n in new_names]})
        print("Updated JIRA %s components to: %s" % (issue.key, ", ".join(new_names)))
    except Exception as e:
        print_error("Failed to update components on JIRA %s: %s" % (issue.key, e))


def get_jira_issue(prompt, default_jira_id=""):
    jira_id = bold_input("%s [%s]: " % (prompt, default_jira_id))
    if jira_id == "":
        jira_id = default_jira_id
        if jira_id == "":
            print("JIRA ID not found, skipping.")
            return None
    try:
        issue = asf_jira.issue(jira_id)
        print_jira_issue_summary(issue)
        status = issue.fields.status.name
        if status == "Resolved" or status == "Closed":
            print("JIRA issue %s already has status '%s'" % (jira_id, status))
            return None
        if get_input("Check if the JIRA information is as expected (y/N): ", ["y", "n", ""]) == "y":
            return issue
        else:
            return get_jira_issue("Enter the revised JIRA ID again or leave blank to skip")
    except Exception as e:
        print_error("ASF JIRA could not find %s: %s" % (jira_id, e))
        return get_jira_issue("Enter the revised JIRA ID again or leave blank to skip")


def resolve_jira_issue(merge_branches, comment, default_jira_id="", title_components=()):
    issue = get_jira_issue("Enter a JIRA id", default_jira_id)
    if issue is None:
        return

    if issue.fields.assignee is None:
        choose_jira_assignee(issue)

    reconcile_jira_components(issue, title_components)

    versions = asf_jira.project_versions("SPARK")
    # Consider only x.y.z, unreleased, unarchived versions
    versions = [
        x
        for x in versions
        if not x.raw["released"] and not x.raw["archived"] and re.match(r"\d+\.\d+\.\d+", x.name)
    ]
    versions = sorted(versions, key=lambda x: x.name, reverse=True)

    unreleased_names = [v.name for v in versions]
    default_fix_list, infer_warnings = compute_merge_default_fix_versions(
        merge_branches, unreleased_names
    )
    for w in infer_warnings:
        print_error(w)
    default_fix_versions = ",".join(default_fix_list)

    available_versions = set(list(map(lambda v: v.name, versions)))
    while True:
        try:
            fix_versions = bold_input(
                "Enter comma-separated fix version(s) [%s]: " % default_fix_versions
            )
            if fix_versions == "":
                fix_versions = default_fix_versions
            fix_versions = fix_versions.replace(" ", "").split(",")
            if set(fix_versions).issubset(available_versions):
                break
            else:
                print(
                    "Specified version(s) [%s] not found in the available versions, try "
                    "again (or leave blank and fix manually)." % (", ".join(fix_versions))
                )
        except KeyboardInterrupt:
            raise
        except BaseException:
            traceback.print_exc()
            print("Error setting fix version(s), try again (or leave blank and fix manually)")

    def get_version_json(version_str):
        return list(filter(lambda v: v.name == version_str, versions))[0].raw

    jira_fix_versions = list(map(lambda v: get_version_json(v), fix_versions))

    resolve = list(filter(lambda a: a["name"] == "Resolve Issue", asf_jira.transitions(issue.key)))[
        0
    ]
    resolution = list(filter(lambda r: r.raw["name"] == "Fixed", asf_jira.resolutions()))[0]
    asf_jira.transition_issue(
        issue.key,
        resolve["id"],
        fixVersions=jira_fix_versions,
        comment=comment,
        resolution={"id": resolution.raw["id"]},
    )

    try:
        print_jira_issue_summary(asf_jira.issue(issue.key))
    except Exception:
        print("Unable to fetch JIRA issue %s after resolving" % issue.key)
    print("Successfully resolved %s with fixVersions=%s!" % (issue.key, fix_versions))


def choose_jira_assignee(issue):
    """
    Prompt the user to choose who to assign the issue to in jira, given a list of candidates,
    including the original reporter and all commentators
    """
    while True:
        try:
            reporter = issue.fields.reporter
            commentators = list(map(lambda x: x.author, issue.fields.comment.comments))
            candidates = set(commentators)
            candidates.add(reporter)
            candidates = list(candidates)
            print("JIRA is unassigned, choose assignee")
            for idx, author in enumerate(candidates):
                if author.key == "apachespark":
                    continue
                annotations = ["Reporter"] if author == reporter else []
                if author in commentators:
                    annotations.append("Commentator")
                print("[%d] %s (%s)" % (idx, author.displayName, ",".join(annotations)))
            raw_assignee = bold_input(
                "Enter number of user, or userid, to assign to (blank to leave unassigned): "
            )
            if raw_assignee == "":
                return None
            else:
                try:
                    id = int(raw_assignee)
                    assignee = candidates[id]
                except BaseException:
                    # assume it's a user id, and try to assign (might fail, we just prompt again)
                    assignee = asf_jira.user(raw_assignee)
                try:
                    assign_issue(issue.key, assignee.name)
                except Exception as e:
                    if (
                        e.__class__.__name__ == "JIRAError"
                        and ("'%s' cannot be assigned" % assignee.name)
                        in getattr(e, "response").text
                    ):
                        continue_maybe(
                            "User '%s' cannot be assigned, add to contributors role and try again?"
                            % assignee.name
                        )
                        grant_contributor_role(assignee.name)
                        assign_issue(issue.key, assignee.name)
                    else:
                        raise e
                return assignee
        except KeyboardInterrupt:
            raise
        except BaseException:
            traceback.print_exc()
            print("Error assigning JIRA, try again (or leave blank and fix manually)")


def grant_contributor_role(user: str):
    role = asf_jira.project_role("SPARK", 10010)
    role.add_user(user)
    print("Successfully added user '%s' to contributors role" % user)


def assign_issue(issue: int, assignee: str) -> bool:
    """
    Assign an issue to a user, which is a shorthand for jira.client.JIRA.assign_issue.
    The original one has an issue that it will search users again and only choose the assignee
    from 20 candidates. If it's unmatched, it picks the head blindly. In our case, the assignee
    is already resolved.
    """
    url = getattr(asf_jira, "_get_latest_url")(f"issue/{issue}/assignee")
    payload = {"name": assignee}
    getattr(asf_jira, "_session").put(url, data=json.dumps(payload))
    return True


def resolve_jira_issues(title, merge_branches, comment, title_components=()):
    jira_ids = re.findall("SPARK-[0-9]{4,5}", title)

    if len(jira_ids) == 0:
        resolve_jira_issue(merge_branches, comment, title_components=title_components)
    for jira_id in jira_ids:
        resolve_jira_issue(merge_branches, comment, jira_id, title_components=title_components)


class Component:
    """A Spark PR-title tag, paired with its canonical JIRA component name.

    ``jira_name`` is the canonical name of the SPARK JIRA component (e.g.
    "Documentation"); empty for status markers like [MINOR] that are not
    JIRA components but are still recognized in PR titles.

    ``tag`` is the preferred PR-title abbreviation (uppercase, no brackets,
    e.g. "DOC"). ``aliases`` lists other accepted spellings that resolve to
    the same component (e.g. "DOCS", "DOCUMENTATION" -> "DOC").

    ``primary`` marks components whose presence alone satisfies the merge-time
    requirement. Non-primary JIRA components (e.g. [TEST], [SHUFFLE], [DEPLOY])
    remain recognized — they normalize and pass through validation — but
    they must be paired with a primary tag (e.g. [SQL][TEST]). Status
    markers are never primary. [WIP] is intentionally absent from the
    registry: a WIP PR should be aborted at the earlier WIP warning, not
    merged.
    """

    def __init__(self, tag, aliases=(), primary=False, jira_name=""):
        self.tag = tag
        self.aliases = frozenset(aliases)
        self.primary = primary
        self.jira_name = jira_name

    def matches(self, token):
        return token == self.tag or token in self.aliases

    @classmethod
    def find(cls, token):
        """Return the Component matching ``token`` (case-insensitive), or None."""
        if token is None:
            return None
        token = token.strip().upper()
        for c in COMPONENTS:
            if c.matches(token):
                return c
        return None


# Full SPARK JIRA component list (sorted alphabetically by tag), followed
# by status markers. Keep in sync with the components in JIRA — fetch the
# current list with:
#   curl -s https://issues.apache.org/jira/rest/api/2/project/SPARK/components
# A `primary=True` marker indicates the tag alone satisfies the merge-time
# component requirement; non-primary JIRA components must be paired with a
# primary one (e.g. [SQL][TEST], [CORE][SHUFFLE]). Status
# markers leave `jira_name` empty.
COMPONENTS = (
    Component("BLOCK_MANAGER", jira_name="Block Manager"),
    Component("BUILD", primary=True, jira_name="Build"),
    Component("CONNECT", primary=True, jira_name="Connect"),
    Component("CORE", ("SPARK_CORE",), primary=True, jira_name="Spark Core"),
    Component("DEPLOY", jira_name="Deploy"),
    Component("DOC", ("DOCS", "DOCUMENTATION"), primary=True, jira_name="Documentation"),
    Component("DOCKER", primary=True, jira_name="Spark Docker"),
    Component("EC2", jira_name="EC2"),
    Component("EXAMPLES", ("EXAMPLE",), primary=True, jira_name="Examples"),
    Component("GRAPHX", primary=True, jira_name="GraphX"),
    Component("INFRA", ("PROJECT_INFRA",), primary=True, jira_name="Project Infra"),
    Component("IO", jira_name="Input/Output"),
    Component("JAVA", ("JAVA_API", "JAVAAPI"), jira_name="Java API"),
    Component("K8S", ("KUBERNETES",), primary=True, jira_name="Kubernetes"),
    Component("MESOS", jira_name="Mesos"),
    Component("ML", primary=True, jira_name="ML"),
    Component("MLLIB", primary=True, jira_name="MLlib"),
    Component("OPTIMIZER", jira_name="Optimizer"),
    Component("PROTOBUF", jira_name="Protobuf"),
    Component("PS", primary=True, jira_name="Pandas API on Spark"),
    Component("PYTHON", ("PYSPARK",), primary=True, jira_name="PySpark"),
    Component("R", ("SPARKR",), primary=True, jira_name="R"),
    Component("REPL", ("SHELL", "SPARK_SHELL"), jira_name="Spark Shell"),
    Component("SCHEDULER", jira_name="Scheduler"),
    Component("SDP", ("PIPELINES",), primary=True, jira_name="Declarative Pipelines"),
    Component("SECURITY", primary=True, jira_name="Security"),
    Component("SHUFFLE", jira_name="Shuffle"),
    Component("SQL", primary=True, jira_name="SQL"),
    Component("SS", primary=True, jira_name="Structured Streaming"),
    Component("STREAMING", ("DSTREAM", "DSTREAMS"), primary=True, jira_name="DStreams"),
    Component("SUBMIT", jira_name="Spark Submit"),
    Component("TEST", ("TESTS", "TEST-ONLY", "TESTS-ONLY"), jira_name="Tests"),
    Component("UI", ("WEBUI", "WEB_UI"), primary=True, jira_name="Web UI"),
    Component("WINDOWS", primary=True, jira_name="Windows"),
    Component("YARN", primary=True, jira_name="YARN"),
    # Status markers — recognized in PR titles, but not JIRA components.
    Component("FOLLOWUP", ("FOLLOW-UP",)),
    Component("MINOR"),
    Component("TRIVIAL"),
)


_BRACKET_TAG_RE = re.compile(r"\[\s*([A-Za-z0-9._-]+)\s*\]")
_SPARK_ID_RE = re.compile(r"^SPARK-\d+$", re.IGNORECASE)
_VERSION_TAG_RE = re.compile(r"^\d+\.(\d+|X)$")
_LEADING_TAGS = frozenset({"MINOR", "TRIVIAL"})


class Title:
    """Structured PR title: SPARK refs, component tags, and body.

    ``leading``    — SPARK-NNNNN IDs and [MINOR]/[TRIVIAL] markers, in order.
    ``components`` — all other bracket tags, in order.
    ``text``       — body text following the bracket sequence.

    >>> t = Title.parse("[SPARK-1234][SQL] Fix something")
    >>> t.leading, t.components, t.text
    (['SPARK-1234'], ['SQL'], 'Fix something')
    >>> str(t)
    '[SPARK-1234][SQL] Fix something'
    >>> t = Title.parse("[SPARK-1234][SQL][FOLLOWUP] Fix something")
    >>> t.leading, t.components, t.text
    (['SPARK-1234'], ['SQL', 'FOLLOWUP'], 'Fix something')
    >>> str(t)
    '[SPARK-1234][SQL][FOLLOWUP] Fix something'
    >>> t = Title.parse("[SPARK-1234]")
    >>> t.leading, t.components, t.text
    (['SPARK-1234'], [], '')
    >>> str(t)
    '[SPARK-1234]'
    """

    def __init__(
        self,
        leading: List[str],
        components: List[str],
        text: str,
    ) -> None:
        self.leading = leading
        self.components = components
        self.text = text

    @classmethod
    def parse(cls, raw: str) -> "Title":
        """Parse a PR title string into a :class:`Title`.

        A title must open with a leading tag ([SPARK-NNNNN], [MINOR], or
        [TRIVIAL]); otherwise :exc:`ValueError` is raised.  Subsequent bracket
        tokens (spaces trimmed, separated by optional whitespace) go to
        ``components``.  The remainder is ``text``.

        >>> t = Title.parse("[SPARK-1234][SQL][TESTS] Fix something")
        >>> t.leading, t.components, t.text
        (['SPARK-1234'], ['SQL', 'TESTS'], 'Fix something')
        >>> t = Title.parse("  [ SPARK-1234 ]  [ SQL ] [  TESTS  ]   Fix something")
        >>> t.leading, t.components, t.text
        (['SPARK-1234'], ['SQL', 'TESTS'], 'Fix something')
        >>> t = Title.parse("[SPARK-1234 ][ sql ][ followup ] Fix")
        >>> t.leading, t.components, t.text
        (['SPARK-1234'], ['SQL', 'FOLLOWUP'], 'Fix')
        >>> str(t)
        '[SPARK-1234][SQL][FOLLOWUP] Fix'
        >>> Title.parse("[MINOR] Fix typo").leading
        ['MINOR']
        >>> t = Title.parse("[spark-1234][sql][followup] Fix")
        >>> t.leading, t.components
        (['SPARK-1234'], ['SQL', 'FOLLOWUP'])
        >>> Title.parse("[SPARK-1234][SPARK-5678][SQL] Fix").leading
        ['SPARK-1234', 'SPARK-5678']
        >>> Title.parse("[SPARK-1234][4.X][SQL] Fix").components
        ['4.X', 'SQL']
        >>> Title.parse("[SPARK-1234][SQL][4.2] Fix").components
        ['SQL', '4.2']
        >>> Title.parse("[SQL] Fix")
        Traceback (most recent call last):
            ...
        ValueError: title must start with [SPARK-NNNNN], [MINOR], or [TRIVIAL]: '[SQL] Fix'
        >>> Title.parse("No brackets")
        Traceback (most recent call last):
            ...
        ValueError: title must start with [SPARK-NNNNN], [MINOR], or [TRIVIAL]: 'No brackets'
        >>> Title.parse("[SPARK-1234][SQL][SPARK-123] Fix")
        Traceback (most recent call last):
            ...
        ValueError: [SPARK-NNNNN] tags must all appear before other tags: '[SPARK-1234][SQL][SPARK-123] Fix'
        >>> Title.parse("[SPARK-1234][MINOR][SQL] Fix")
        Traceback (most recent call last):
            ...
        ValueError: [SPARK-NNNNN], [MINOR], and [TRIVIAL] cannot coexist
        >>> Title.parse("[MINOR][TRIVIAL][SQL] Fix")
        Traceback (most recent call last):
            ...
        ValueError: [SPARK-NNNNN], [MINOR], and [TRIVIAL] cannot coexist
        """
        leading: List[str] = []
        components: List[str] = []

        raw = raw.strip()
        m0 = _BRACKET_TAG_RE.match(raw)
        first = m0.group(1).upper() if m0 else ""
        if not (_SPARK_ID_RE.match(first) or first in _LEADING_TAGS):
            raise ValueError("title must start with [SPARK-NNNNN], [MINOR], or [TRIVIAL]: %r" % raw)

        past_leading = False
        pos = 0
        while pos < len(raw):
            m = _BRACKET_TAG_RE.match(raw, pos)
            if not m:
                break
            tag = m.group(1).upper()
            if _SPARK_ID_RE.match(tag):
                if past_leading:
                    raise ValueError(
                        "[SPARK-NNNNN] tags must all appear before other tags: %r" % raw
                    )
                leading.append(tag)
            elif tag in _LEADING_TAGS:
                leading.append(tag)
            else:
                components.append(tag)
                past_leading = True
            pos = m.end()
            while pos < len(raw) and raw[pos] == " ":
                pos += 1

        text = raw[pos:].lstrip()
        markers = [t for t in leading if t in _LEADING_TAGS]
        if len(markers) > 1 or (markers and len(leading) > len(markers)):
            raise ValueError("[SPARK-NNNNN], [MINOR], and [TRIVIAL] cannot coexist")
        return cls(leading, components, text)

    def __str__(self) -> str:
        parts = "".join("[%s]" % t for t in self.leading)
        parts += "".join("[%s]" % c for c in self.components)
        if not self.text:
            return parts
        return parts + (" " if parts else "") + self.text


def prompt_for_components():
    """
    Prompt the committer for component(s) when the PR title lacks a primary
    component. Each entered token is normalized via Component.find
    (e.g. "DOCS" -> "DOC", "PYSPARK" -> "PYTHON"). Unrecognized tokens are
    passed through as-is. Re-prompts until at least one entered token resolves
    to a primary Component (one with primary=True). Returns an uppercase list
    of tags in insertion order.
    """
    print("PR title is missing a primary [COMPONENT] tag.")
    print("Primary components (one of these is required):")
    primary = [c for c in COMPONENTS if c.primary]
    width = max(len(c.tag) for c in primary)
    for c in primary:
        print("  [%s]%s - %s" % (c.tag, " " * (width - len(c.tag)), c.jira_name))
    while True:
        raw = bold_input(
            "Enter comma-separated component(s) to insert into the title (e.g. CORE,SQL): "
        )
        components = []
        has_primary = False
        for token in raw.split(","):
            t = token.strip().upper()
            if t:
                c = Component.find(t)
                if c is not None and c.primary:
                    has_primary = True
                components.append(c.tag if c else t)
        if not components:
            print_error("Component(s) cannot be empty. Please enter at least one.")
            continue
        if not has_primary:
            print_error(
                "At least one component must be a primary tag (see list above). "
                "Got: %s" % ", ".join(components)
            )
            continue
        return components


def get_current_ref():
    ref = run_cmd("git rev-parse --abbrev-ref HEAD").strip()
    if ref == "HEAD":
        # The current ref is a detached HEAD, so grab its SHA.
        return run_cmd("git rev-parse HEAD").strip()
    else:
        return ref


def initialize_jira():
    # JIRA access is mandatory: a successful merge must resolve the associated ticket. Bail out
    # early -- before any git work -- if a prerequisite is missing, rather than silently merging
    # and leaving the JIRA open (which is easy to miss and tedious to reconcile after the fact).
    global asf_jira
    asf_jira = None
    jira_server = {"server": JIRA_API_BASE}

    if not JIRA_IMPORTED:
        fail("ERROR finding jira library. Run 'pip3 install jira' to install, then retry.")
    elif JIRA_ACCESS_TOKEN:
        client = jira.client.JIRA(
            jira_server, token_auth=JIRA_ACCESS_TOKEN, timeout=(JIRA_CONNECT_TIMEOUT, 30)
        )
        try:
            # Eagerly check if the token is valid to align with the behavior of username/password
            # authn
            client.current_user()
            asf_jira = client
        except Exception as e:
            if e.__class__.__name__ == "JIRAError" and getattr(e, "status_code", None) == 401:
                msg = (
                    "ASF JIRA could not authenticate with the invalid or expired token '%s'"
                    % JIRA_ACCESS_TOKEN
                )
                fail(msg)
            else:
                raise e
    elif JIRA_USERNAME and JIRA_PASSWORD:
        print("You can use JIRA_ACCESS_TOKEN instead of JIRA_USERNAME/JIRA_PASSWORD.")
        print("Visit https://issues.apache.org/jira/secure/ViewProfile.jspa ")
        print("and click 'Personal Access Tokens' menu to manage your own tokens.")
        asf_jira = jira.client.JIRA(
            jira_server,
            basic_auth=(JIRA_USERNAME, JIRA_PASSWORD),
            timeout=(JIRA_CONNECT_TIMEOUT, 30),
        )
    else:
        fail(
            "Neither JIRA_ACCESS_TOKEN nor JIRA_USERNAME/JIRA_PASSWORD are set. Configure JIRA "
            "credentials (a personal access token is recommended) and retry. See "
            "https://issues.apache.org/jira/secure/ViewProfile.jspa -> Personal Access Tokens."
        )


def check_script_up_to_date():
    # Running a stale merge script can silently reintroduce already-fixed behavior (e.g. skipping
    # JIRA resolution). Rather than download the whole file, ask GitHub only for the latest commit
    # on master that touched this script (a small response) and fail if that commit is not already
    # in the local history -- i.e. the committer is missing an upstream update. Being ahead of
    # master (branched from it) is fine and does not trip the check. Best-effort: if master cannot
    # be reached we warn and proceed rather than block merges on a transient network/rate error.
    if SKIP_VERSION_CHECK:
        print("SKIP_VERSION_CHECK is set; not verifying this script against master.")
        return

    url = "%s/commits?path=%s&sha=master&per_page=1" % (GITHUB_API_BASE, MERGE_SCRIPT_REPO_PATH)
    try:
        request = Request(url)
        if GITHUB_OAUTH_KEY:
            request.add_header("Authorization", "token %s" % GITHUB_OAUTH_KEY)
        commits = json.load(urlopen(request))
    except Exception as e:
        print_error(
            "Could not check %s against master (%s). Proceeding with the local copy."
            % (MERGE_SCRIPT_REPO_PATH, e)
        )
        return
    if not commits:
        return
    latest_sha = commits[0]["sha"]

    # `git merge-base --is-ancestor` exits 0 when latest_sha is reachable from HEAD (we have it),
    # nonzero when it is missing (stale) or unknown locally (never fetched).
    try:
        subprocess.check_call(
            ["git", "merge-base", "--is-ancestor", latest_sha, "HEAD"],
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        return
    except subprocess.CalledProcessError:
        pass

    fail(
        "Your local %s is out of date: apache/spark master has a newer revision (%s) that is not "
        "in your local history. Update it before merging, e.g.:\n"
        "    git fetch %s master && git checkout %s/master -- %s\n"
        "then re-run. Set SKIP_VERSION_CHECK=1 to bypass (e.g. when changing this script itself)."
        % (
            MERGE_SCRIPT_REPO_PATH,
            latest_sha[:12],
            PUSH_REMOTE_NAME,
            PUSH_REMOTE_NAME,
            MERGE_SCRIPT_REPO_PATH,
        )
    )


def main():
    check_script_up_to_date()
    initialize_jira()
    global original_head

    os.chdir(SPARK_HOME)
    original_head = get_current_ref()

    branches = get_json("%s/branches" % GITHUB_API_BASE)
    branch_names = list(filter(lambda x: x.startswith("branch-"), [x["name"] for x in branches]))
    branch_names = sorted(branch_names, key=semver_branch_rank, reverse=True)

    if len(sys.argv) == 1:
        pr_num = get_input("Which pull request would you like to merge? (e.g. 34): ", r"^\d+$")
    else:
        pr_num = sys.argv[1]
        print("Start to merge pull request #%s" % (pr_num))
    pr = get_json("%s/pulls/%s" % (GITHUB_API_BASE, pr_num))
    pr_events = get_json("%s/issues/%s/events" % (GITHUB_API_BASE, pr_num))

    url = pr["url"]
    title = pr["title"]

    # Fail hard on draft, WIP, or DO-NOT-MERGE PRs to prevent accidental merges.
    if pr.get("draft", False) or "[WIP]" in title or "[DO-NOT-MERGE]" in title:
        fail("Cannot merge a draft PR #%s: %s" % (pr_num, title))

    # e.g. 'Revert "[SPARK-56357][BUILD] Upgrade sbt to 1.12.8"'
    is_revert_pr = title.startswith('Revert "') and title.endswith('"')
    # e.g. 'Reapply "[SPARK-56357][BUILD] Upgrade sbt to 1.12.8"'
    is_reapply_pr = title.startswith('Reapply "') and title.endswith('"')

    # Normalized PR-title component tags, used later to reconcile JIRA components. Empty for
    # Revert/Reapply PRs, whose titles are kept verbatim and not parsed for components.
    title_components: List[str] = []

    # Revert and Reapply PRs keep their title verbatim.
    if not (is_revert_pr or is_reapply_pr):
        # Parse; fail on a malformed title.
        try:
            parsed = Title.parse(title)
        except ValueError as e:
            fail("Malformed PR title: %s" % e)

        # Normalize component tags via the registry and track primary.
        components = []
        has_primary = False
        for tag in parsed.components:
            c = Component.find(tag)
            if c is not None and c.primary:
                has_primary = True
            components.append(c.tag if c is not None else tag)
        if not has_primary:
            new_tags = prompt_for_components()
            components = new_tags + components

        # Deduplicate tags in insertion order.
        components = list(dict.fromkeys(components))

        # Move version tags (e.g. [4.X], [4.2]) to the head of components.
        versions = [t for t in components if _VERSION_TAG_RE.match(t)]
        if versions:
            others = [t for t in components if not _VERSION_TAG_RE.match(t)]
            components = versions + others

        # Move FOLLOWUP to the last tag.
        non_followup = [t for t in components if t != "FOLLOWUP"]
        if len(non_followup) < len(components):
            components = non_followup + ["FOLLOWUP"]

        # Warn about tags that are neither known components nor version tags.
        unknown = [
            t for t in components if Component.find(t) is None and not _VERSION_TAG_RE.match(t)
        ]
        if unknown:
            print_error("Title has unknown tag(s): %s" % ", ".join("[%s]" % t for t in unknown))

        parsed.components = components
        title_components = list(parsed.components)
        title = str(parsed)
        if title != pr["title"]:
            print("Normalized title: %s" % title)

    body = pr["body"]
    if body is None:
        body = ""
    modified_body = re.sub(re.compile(r"<!--[^>]*-->\n?", re.DOTALL), "", body).lstrip()
    if modified_body != body:
        print("=" * 80)
        print(modified_body)
        print("=" * 80)
        print("I've removed the comments from PR template like the above:")
        if get_input("Would you like to use the modified body? (y/N): ", ["y", "n", ""]) == "y":
            body = modified_body
            print("Using modified body:")
        else:
            print("Using original body:")
        print("=" * 80)
        print(body)
        print("=" * 80)
    target_ref = pr["base"]["ref"]
    user_login = pr["user"]["login"]
    base_ref = pr["head"]["ref"]
    pr_repo_desc = "%s/%s" % (user_login, base_ref)

    # Fetch PR author's GitHub profile and PR commits for commit attribution
    pr_author_info = get_json("https://api.github.com/users/%s" % user_login)
    pr_author_name = pr_author_info.get("name") or user_login
    pr_author_email = pr_author_info.get("email")
    pr_commits = get_json("%s/pulls/%s/commits" % (GITHUB_API_BASE, pr_num))
    if not pr_author_email:
        # If the GitHub profile has no public email, find the author's email from the
        # PR's git commits. Only use it if GitHub links the commit to the PR author's
        # account (meaning the email is verified on their GitHub account).
        for c in pr_commits:
            commit_author = c.get("author")
            if commit_author and commit_author.get("login") == user_login:
                pr_author_email = c["commit"]["author"]["email"]
                break
    if not pr_author_email:
        pr_author_email = "%s+%s@users.noreply.github.com" % (pr_author_info["id"], user_login)
    pr_author = "%s <%s>" % (pr_author_name, pr_author_email)

    # Collect co-authors: commit authors whose GitHub login differs from the PR author.
    # This deduplicates the PR author (who may use multiple emails across commits) by login.
    co_authors = []
    seen = set()
    for c in pr_commits:
        gh_author = c.get("author")
        if gh_author and gh_author.get("login") == user_login:
            continue
        raw = "%s <%s>" % (c["commit"]["author"]["name"], c["commit"]["author"]["email"])
        if raw not in seen:
            seen.add(raw)
            co_authors.append(raw)

    # Merged pull requests don't appear as merged in the GitHub API;
    # Instead, they're closed by committers.
    # A PR might have multiple merge commits, if it's reopened and merged again. We shall
    # cherry-pick PRs in closed state with the latest merge hash.
    # If the PR is still open (reopened), we shall not cherry-pick it but perform the normal
    # merge as it could have been reverted earlier.
    merge_hash, message = (None, None)
    if pr["state"] == "closed":
        merge_hash, message = find_merge_commit(pr_num, pr_events)

    if merge_hash is not None:
        print("Pull request %s has already been merged, assuming you want to backport" % pr_num)
        commit_is_downloaded = (
            run_cmd(["git", "rev-parse", "--quiet", "--verify", "%s^{commit}" % merge_hash]).strip()
            != ""
        )
        if not commit_is_downloaded:
            fail("Couldn't find any merge commit for #%s, you may need to update HEAD." % pr_num)

        print("Found commit %s:\n%s" % (merge_hash, message))
        # The change is already on target_ref and on any branch a previous run backported it
        # to, so exclude all of them: defaulting to one would cherry-pick an empty commit.
        picked_refs = [target_ref] + [
            b for b in branches_with_merge_footer(pr_num, branch_names) if b != target_ref
        ]
        if len(picked_refs) > 1:
            print("Already backported to: %s" % ", ".join(picked_refs[1:]))
        # Loop so one invocation can reach several maintenance branches, as the merge path does.
        picked_commits = []
        try:
            while True:
                default = default_pick_branch(branch_names, tuple(picked_refs))
                if default is None:
                    print(
                        "Every known release branch already contains #%s; nothing to pick." % pr_num
                    )
                    break
                picked = cherry_pick(
                    pr_num,
                    merge_hash,
                    default,
                    branch_names,
                    target_ref,
                    already_picked=tuple(picked_refs),
                )
                picked_refs = picked_refs + [ref for ref, _ in picked]
                picked_commits = picked_commits + picked
                prompt = "Would you like to pick %s into another branch?" % merge_hash
                if get_input(f"\n{prompt} (y/N): ", ["y", "n", ""]) != "y":
                    break
        finally:
            # Report whatever was pushed even if a later pick is aborted, since the earlier
            # pushes have already landed.
            if picked_commits:
                post_merge_comment(pr_num, picked_commits)
        sys.exit(0)

    if not bool(pr["mergeable"]):
        msg = (
            "Pull request %s is not mergeable in its current form.\n" % pr_num
            + "Continue? (experts only!)"
        )
        continue_maybe(msg)

    # asf_jira is guaranteed to be set here: initialize_jira() fails fast otherwise.
    jira_ids = re.findall("SPARK-[0-9]{4,5}", title)
    # Epic / Umbrella tickets group related work and must not be resolved by a single PR.
    # Collect every offender so the committer sees the full list in one shot rather than
    # discovering them one-by-one across repeated merge attempts.
    blocking_issue_types = {"Epic", "Umbrella"}
    blockers = []
    linked_issues = []
    for jira_id in jira_ids:
        try:
            issue = asf_jira.issue(jira_id)
        except Exception:
            print_error("Unable to fetch summary of %s" % jira_id)
            continue
        print_jira_issue_summary(issue)
        linked_issues.append((jira_id, issue))
        issue_type = issue.fields.issuetype.name
        if issue_type in blocking_issue_types:
            blockers.append((jira_id, issue_type))
    if blockers:
        ids_str = ", ".join("%s (%s)" % (jid, t) for jid, t in blockers)
        fail(
            "Cannot merge PR #%s. Linked JIRA(s) %s are Umbrella or Epic "
            "tickets and MUST not be resolved by a single PR. File "
            "Sub-task(s) under %s and update the PR title to reference "
            "the Sub-task(s) instead." % (pr_num, ids_str, ids_str)
        )

    # Confirm each linked JIRA actually matches this PR before merging. A committer can
    # reference the wrong ticket -- e.g. mistaking a GitHub PR number in the body
    # ("Closes #NNNNN") for a SPARK JIRA id -- which resolves an unrelated ticket on merge.
    # Show the PR title next to each ticket's summary, warn when a ticket is already
    # Resolved/Closed (a fresh merge should target an open one), and require confirmation.
    # For [FOLLOWUP] PRs the title describes the follow-up fix rather than the ticket, so
    # its low similarity to the JIRA summary is expected and the warning is suppressed.
    is_followup = "FOLLOWUP" in title_components
    for jira_id, issue in linked_issues:
        print()
        print(
            format_jira_verification(
                pr_num,
                title,
                jira_id,
                issue.fields.summary,
                issue.fields.status.name,
                issue.fields.issuetype.name,
                use_color=True,
                is_followup=is_followup,
            )
        )
        if get_input("Does %s match this PR? (y/N): " % jira_id, ["y", "n", ""]) != "y":
            fail(
                "Aborting: %s does not match PR #%s. Fix the PR title to reference the "
                "correct JIRA ticket and retry." % (jira_id, pr_num)
            )

    print("\n=== Pull Request #%s ===" % pr_num)
    print("title\t%s\nsource\t%s\ntarget\t%s\nurl\t%s" % (title, pr_repo_desc, target_ref, url))
    continue_maybe("Proceed with merging pull request #%s?" % pr_num)

    merged_refs = [target_ref]

    merge_hash = merge_pr(pr_num, target_ref, title, body, pr_repo_desc, pr_author, co_authors)

    # Ordered (branch, commit_hash) pairs for the merge comment: the merge sink first,
    # then each cherry-pick target as it is picked.
    merged_commits = [(target_ref, merge_hash)]

    # merged_refs drives both the next prompt default and the already_picked set passed to
    # cherry_pick, so each grows with every cherry-pick -- including branches consumed by the
    # Upstream-First two-branch path inside cherry_pick (e.g. picking branch-M.x + branch-M.N
    # in a single prompt). It starts with target_ref, the merge sink, never to be re-picked.
    pick_prompt = "Would you like to pick %s into another branch?" % merge_hash
    # Always record the merge summary for what actually landed, even if a later
    # cherry-pick is aborted or cancelled: the merge into the target branch has
    # already been pushed, so cancelling a backport must not drop that line.
    try:
        while get_input(f"\n{pick_prompt} (y/N): ", ["y", "n", ""]) == "y":
            default = default_pick_branch(branch_names, tuple(merged_refs))
            if default is None:
                print("Every known release branch already contains #%s; nothing to pick." % pr_num)
                break
            picked = cherry_pick(
                pr_num,
                merge_hash,
                default,
                branch_names,
                target_ref,
                already_picked=tuple(merged_refs),
            )
            merged_refs = merged_refs + [ref for ref, _ in picked]
            merged_commits = merged_commits + picked
    finally:
        if merged_commits:
            # The "Closes #N" keyword in the commit message only auto-closes the PR when the
            # commit lands on the default branch. For merges into other branches (e.g.
            # branch-X.Y backport PRs), GitHub leaves the PR open, so close it through the API.
            pr_state = get_json("%s/pulls/%s" % (GITHUB_API_BASE, pr_num)).get("state")
            if pr_state != "closed":
                close_message = "PR #%s is still open after push; closing it explicitly." % pr_num
                print("\n%s\n" % bold(close_message))
                close_pr(pr_num)
            # Record every branch that successfully received the change on the PR.
            post_merge_comment(pr_num, merged_commits)

    # asf_jira is guaranteed to be set here: initialize_jira() fails fast otherwise.
    continue_maybe("Would you like to update an associated JIRA?")
    jira_comment = "Issue resolved by pull request %s\n[%s/%s]" % (
        pr_num,
        GITHUB_BASE,
        pr_num,
    )
    resolve_jira_issues(title, merged_refs, jira_comment, title_components)


if __name__ == "__main__":
    import doctest

    failure_count, test_count = doctest.testmod()
    if failure_count:
        sys.exit(-1)
    try:
        main()
    except BaseException:
        clean_up()
        raise