def METHOD_NAME(
    self, message: Mapping[str, Any], metadata: KafkaMessageMetadata
) -> Optional[ProcessedMessage]:
    # TODO: Support messages with multiple buckets
    if not self._should_process(message):
        return None
    timestamp = _ensure_valid_date(datetime.utcfromtimestamp(message["timestamp"]))
    assert timestamp is not None, "Invalid timestamp"
    keys = []
    values = []
    tags = message["tags"]
    assert isinstance(tags, Mapping), "Invalid tags type"
    sorted_tag_items = sorted(tags.items())
    for key, value in sorted_tag_items:
        assert key.isdigit() and isinstance(value, int), "Tag key/value invalid"
        keys.append(int(key))
        values.append(value)
    mat_version = (
        DISABLED_MATERIALIZATION_VERSION
        if settings.WRITE_METRICS_AGG_DIRECTLY
        else settings.ENABLED_MATERIALIZATION_VERSION
    )
    try:
        retention_days = enforce_retention(message["retention_days"], timestamp)
    except EventTooOld:
        return None
    processed = {
        "org_id": message["org_id"],
        "project_id": message["project_id"],
        "metric_id": message["metric_id"],
        "timestamp": timestamp,
        "tags.key": keys,
        "tags.value": values,
        **self._process_values(message),
        "materialization_version": mat_version,
        "retention_days": retention_days,
        "timeseries_id": self._hash_timeseries_id(message, sorted_tag_items),
        "partition": metadata.partition,
        "offset": metadata.offset,
    }
    sentry_received_timestamp = None
    if message.get("sentry_received_timestamp"):
        sentry_received_timestamp = datetime.utcfromtimestamp(
            message["sentry_received_timestamp"]
        )
    return InsertBatch(
        [processed], None, sentry_received_timestamp=sentry_received_timestamp
    )