def METHOD_NAME(
    headers: Sequence[Tuple[str, Optional[bytes]]]
) -> Optional[dict[str, Any]]:
    """
    Same as get_task_kwargs_for_message but gets the required information from
    the kafka message headers.
    """
    try:
        header_data = {k: v for k, v in headers}
        def _required(k: str) -> bytes:
            v = header_data[k]
            if v is None:
                raise ValueError(f"expected {k!r} to be non-None")
            return v
        version = decode_int(_required("version"))
        operation = decode_str(_required("operation"))
        if operation == "insert":
            if "group_id" not in header_data:
                header_data["group_id"] = None
            if "primary_hash" not in header_data:
                header_data["primary_hash"] = None
            primary_hash = decode_optional_str(header_data["primary_hash"])
            event_id = decode_str(_required("event_id"))
            group_id = decode_optional_int(header_data["group_id"])
            project_id = decode_int(_required("project_id"))
            event_data = {
                "event_id": event_id,
                "group_id": group_id,
                "project_id": project_id,
                "primary_hash": primary_hash,
                "occurrence_id": decode_optional_str(header_data.get("occurrence_id")),
            }
            skip_consume = decode_bool(_required("skip_consume"))
            is_new = decode_bool(_required("is_new"))
            is_regression = decode_bool(_required("is_regression"))
            is_new_group_environment = decode_bool(_required("is_new_group_environment"))
            task_state: MutableMapping[str, Any] = {
                "skip_consume": skip_consume,
                "is_new": is_new,
                "is_regression": is_regression,
                "is_new_group_environment": is_new_group_environment,
            }
            group_states_str = decode_optional_str(header_data.get("group_states"))
            group_states = None
            try:
                group_states = decode_optional_list_str(group_states_str)
            except ValueError:
                logger.error(f"Received event with malformed group_states: '{group_states_str}'")
            except Exception:
                logger.error(
                    f"Uncaught exception thrown when trying to parse group_states: '{group_states_str}'"
                )
            task_state["group_states"] = group_states
            # default in case queue is not sent
            task_state["queue"] = (
                decode_str(_required("queue")) if "queue" in header_data else "post_process_errors"
            )
        else:
            event_data = {}
            task_state = {}
    except Exception:
        raise InvalidPayload("Received event payload with unexpected structure")
    try:
        handler = version_handlers[version]
    except (ValueError, KeyError):
        raise InvalidVersion(
            f"Received event payload with unexpected version identifier: {version}"
        )
    return handler(operation, event_data, task_state)