def METHOD_NAME(loader_config: dict[str, Any]) -> KafkaStreamLoader:
    processor_name = loader_config["processor"]
    processor = DatasetMessageProcessor.from_name(processor_name)
    assert processor is not None
    default_topic = Topic(loader_config["default_topic"])
    # optionals
    pre_filter = None
    if PRE_FILTER in loader_config and loader_config[PRE_FILTER] is not None:
        pre_filter = StreamMessageFilter.get_from_name(
            loader_config[PRE_FILTER]["type"]
        ).from_kwargs(**loader_config[PRE_FILTER].get("args", {}))
    replacement_topic = __get_topic(loader_config, "replacement_topic")
    commit_log_topic = __get_topic(loader_config, "commit_log_topic")
    subscription_scheduled_topic = __get_topic(
        loader_config, "subscription_scheduled_topic"
    )
    subscription_scheduler_mode = (
        SchedulingWatermarkMode(loader_config[SUBCRIPTION_SCHEDULER_MODE])
        if SUBCRIPTION_SCHEDULER_MODE in loader_config
        and loader_config[SUBCRIPTION_SCHEDULER_MODE] is not None
        else None
    )
    subscription_result_topic = __get_topic(loader_config, "subscription_result_topic")
    dlq_topic = __get_topic(loader_config, "dlq_topic")
    return build_kafka_stream_loader_from_settings(
        processor,
        default_topic,
        pre_filter,
        replacement_topic,
        commit_log_topic,
        subscription_scheduler_mode,
        subscription_scheduled_topic,
        subscription_result_topic,
        dlq_topic,
    )