def METHOD_NAME(
    self,
    outer_message: Message[MessageBatch],
) -> IndexerOutputMessageBatch:
    """
    We have an outer_message Message() whose payload is a batch of Message() objects.
        Message(
            partition=...,
            offset=...
            timestamp=...
            payload=[Message(...), Message(...), etc]
        )
    The inner messages payloads are KafkaPayload's that have:
        * key
        * headers
        * value
    The value of the message is what we need to parse and then translate
    using the indexer.
    """
    should_index_tag_values = self._config.should_index_tag_values
    is_output_sliced = self._config.is_output_sliced or False
    batch = IndexerBatch(
        outer_message,
        should_index_tag_values=should_index_tag_values,
        is_output_sliced=is_output_sliced,
        input_codec=_INGEST_CODEC,
        tags_validator=self.__get_tags_validator(),
    )
    sdk.set_measurement("indexer_batch.payloads.len", len(batch.parsed_payloads_by_offset))
    with metrics.timer("metrics_consumer.check_cardinality_limits"), sentry_sdk.start_span(
        op="check_cardinality_limits"
    ):
        cardinality_limiter = cardinality_limiter_factory.get_ratelimiter(self._config)
        cardinality_limiter_state = cardinality_limiter.check_cardinality_limits(
            self._config.use_case_id, batch.parsed_payloads_by_offset
        )
    sdk.set_measurement(
        "cardinality_limiter.keys_to_remove.len", len(cardinality_limiter_state.keys_to_remove)
    )
    batch.filter_messages(cardinality_limiter_state.keys_to_remove)
    extracted_strings = batch.extract_strings()
    sdk.set_measurement("org_strings.len", len(extracted_strings))
    with metrics.timer("metrics_consumer.bulk_record"), sentry_sdk.start_span(op="bulk_record"):
        record_result = self._indexer.bulk_record(extracted_strings)
    mapping = record_result.get_mapped_results()
    bulk_record_meta = record_result.get_fetch_metadata()
    new_messages = batch.reconstruct_messages(mapping, bulk_record_meta)
    sdk.set_measurement("new_messages.len", len(new_messages))
    with metrics.timer("metrics_consumer.apply_cardinality_limits"), sentry_sdk.start_span(
        op="apply_cardinality_limits"
    ):
        # TODO: move to separate thread
        cardinality_limiter.apply_cardinality_limits(cardinality_limiter_state)
    return new_messages