You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: tests.rs
    path of file: ./repos/quickwit/quickwit/quickwit-control-plane/src
    
    the code of the file until where you have to start completion: // Copyright (C) 2024 Quickwit, Inc.
//
// Quickwit is offered under the AGPL v3.0 and as commercial software.
// For commercial licensing, contact us at hello@quickwit.io.
//
// AGPL:
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as
// published by the Free Software Foundation, either version 3 of the
// License, or (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <http://www.gnu.org/licenses/>.

use std::num::NonZeroUsize;
use std::time::Duration;

use fnv::FnvHashMap;
use futures::{Stream, StreamExt};
use quickwit_actors::{Inbox, Mailbox, Observe, Universe};
use quickwit_cluster::{create_cluster_for_test, ChannelTransport, Cluster, ClusterChange};
use quickwit_common::test_utils::wait_until_predicate;
use quickwit_common::tower::{Change, Pool};
use quickwit_config::service::QuickwitService;
use quickwit_config::{
    ClusterConfig, KafkaSourceParams, SourceConfig, SourceInputFormat, SourceParams,
};
use quickwit_indexing::IndexingService;
use quickwit_metastore::{IndexMetadata, ListIndexesMetadataResponseExt};
use quickwit_proto::indexing::{ApplyIndexingPlanRequest, CpuCapacity, IndexingServiceClient};
use quickwit_proto::metastore::{
    ListIndexesMetadataResponse, ListShardsResponse, MetastoreServiceClient,
};
use quickwit_proto::types::NodeId;
use serde_json::json;

use crate::control_plane::{ControlPlane, CONTROL_PLAN_LOOP_INTERVAL};
use crate::indexing_scheduler::MIN_DURATION_BETWEEN_SCHEDULING;
use crate::IndexerNodeInfo;

fn index_metadata_for_test(
    index_id: &str,
    source_id: &str,
    desired_num_pipelines: usize,
    max_num_pipelines_per_indexer: usize,
) -> IndexMetadata {
    let mut index_metadata = IndexMetadata::for_test(index_id, "ram://indexes/test-index");
    let source_config = SourceConfig {
        enabled: true,
        source_id: source_id.to_string(),
        max_num_pipelines_per_indexer: NonZeroUsize::new(max_num_pipelines_per_indexer).unwrap(),
        desired_num_pipelines: NonZeroUsize::new(desired_num_pipelines).unwrap(),
        source_params: SourceParams::Kafka(KafkaSourceParams {
            topic: "topic".to_string(),
            client_log_level: None,
            client_params: json!({
            "bootstrap.servers": "localhost:9092",
            }),
            enable_backfill_mode: true,
        }),
        transform_config: None,
        input_format: SourceInputFormat::Json,
    };
    index_metadata
        .sources
        .insert(source_id.to_string(), source_config);
    index_metadata
}

pub fn test_indexer_change_stream(
    cluster_change_stream: impl Stream<Item = ClusterChange> + Send + 'static,
    indexing_clients: FnvHashMap<String, Mailbox<IndexingService>>,
) -> impl Stream<Item = Change<String, IndexerNodeInfo>> + Send + 'static {
    cluster_change_stream.filter_map(move |cluster_change| {
        let indexing_clients = indexing_clients.clone();
        Box::pin(async move {
            match cluster_change {
                ClusterChange::Add(node)
                    if node.enabled_services().contains(&QuickwitService::Indexer) =>
                {
                    let node_id = node.node_id().to_string();
                    let generation_id = node.chitchat_id().generation_id;
                    let indexing_tasks = node.indexing_tasks().to_vec();
                    let client_mailbox = indexing_clients.get(&node_id).unwrap().clone();
                    let client = IndexingServiceClient::from_mailbox(client_mailbox);
                    Some(Change::Insert(
                        node_id.clone(),
                        IndexerNodeInfo {
                            node_id: NodeId::from(node_id),
                            
    
async fn test_scheduler_scheduling_multiple_indexers() {
    let transport = ChannelTransport::default();
    let cluster = create_cluster_for_test(Vec::new(), &["control_plane"], &transport, true)
        .await
        .unwrap();
    let cluster_indexer_1 = create_cluster_for_test(
        vec![cluster.gossip_advertise_addr().to_string()],
        &["indexer"],
        &transport,
        true,
    )
    .await
    .unwrap();
    let cluster_indexer_2 = create_cluster_for_test(
        vec![cluster.gossip_advertise_addr().to_string()],
        &["indexer"],
        &transport,
        true,
    )
    .await
    .unwrap();
    let universe = Universe::new();
    let (indexing_service_inboxes, control_plane_mailbox) = start_control_plane(
        cluster.clone(),
        &[&cluster_indexer_1, &cluster_indexer_2],
        &universe,
    )
    .await;
    let indexing_service_inbox_1 = indexing_service_inboxes[0].clone();
    let indexing_service_inbox_2 = indexing_service_inboxes[1].clone();

    // No indexer.
    let scheduler_state = control_plane_mailbox
        .ask(Observe)
        .await
        .unwrap()
        .indexing_scheduler;
    let indexing_service_inbox_messages =
        indexing_service_inbox_1.drain_for_test_typed::<ApplyIndexingPlanRequest>();
    assert_eq!(scheduler_state.num_applied_physical_indexing_plan, 0);
    assert_eq!(scheduler_state.num_schedule_indexing_plan, 0);
    assert!(scheduler_state.last_applied_physical_plan.is_none());
    assert_eq!(indexing_service_inbox_messages.len(), 0);

    cluster
        .wait_for_ready_members(
            |members| {
                members
                    .iter()
                    .any(|member| member.enabled_services.contains(&QuickwitService::Indexer))
            },
            Duration::from_secs(5),
        )
        .await
        .unwrap();

    // Wait for chitchat update, sheduler will detect new indexers and schedule a plan.
    wait_until_predicate(
        || {
            let control_plane_mailbox_clone = control_plane_mailbox.clone();
            async move {
                let scheduler_state = control_plane_mailbox_clone
                    .ask(Observe)
                    .await
                    .unwrap()
                    .indexing_scheduler;
                scheduler_state.num_schedule_indexing_plan == 1
            }
        },
        CONTROL_PLAN_LOOP_INTERVAL * 4,
        Duration::from_millis(100),
    )
    .await
    .unwrap();
    let scheduler_state = control_plane_mailbox
        .ask(Observe)
        .await
        .unwrap()
        .indexing_scheduler;
    assert_eq!(scheduler_state.num_applied_physical_indexing_plan, 1);
    let indexing_service_inbox_messages_1 =
        indexing_service_inbox_1.drain_for_test_typed::<ApplyIndexingPlanRequest>();
    let indexing_service_inbox_messages_2 =
        indexing_service_inbox_2.drain_for_test_typed::<ApplyIndexingPlanRequest>();
    assert_eq!(indexing_service_inbox_messages_1.len(), 1);
    assert_eq!(indexing_service_inbox_messages_2.len(), 1);
    cluster_indexer_1
        .update_self_node_indexing_tasks(&indexing_service_inbox_messages_1[0].indexing_tasks)
        .await
        .unwrap();
    cluster_indexer_2
        .update_self_node_indexing_tasks(&indexing_service_inbox_messages_2[0].indexing_tasks)
        .await
        .unwrap();

    // Wait 2 CONTROL_PLAN_LOOP_INTERVAL again and check the scheduler will not apply the plan
    // several times.
    universe.sleep(CONTROL_PLAN_LOOP_INTERVAL * 2).await;
    let scheduler_state = control_plane_mailbox
        .ask(Observe)
        .await
        .unwrap()
        .indexing_scheduler;
    assert_eq!(scheduler_state.num_schedule_indexing_plan, 1);

    // Shutdown cluster and wait until the new scheduling.
    cluster_indexer_2.shutdown().await;

    cluster
        .wait_for_ready_members(
            |members| {
                members
                    .iter()
                    .filter(|member| member.enabled_services.contains(&QuickwitService::Indexer))
                    .count()
                    == 1
            },
            Duration::from_secs(5),
        )
        .await
        .unwrap();

    wait_until_predicate(
        || {
            let scheduler_handler_mailbox_clone = control_plane_mailbox.clone();
            async move {
                let scheduler_state = scheduler_handler_mailbox_clone
                    .ask(Observe)
                    .await
                    .unwrap()
                    .indexing_scheduler;
                scheduler_state.num_schedule_indexing_plan == 2
            }
        },
        CONTROL_PLAN_LOOP_INTERVAL * 10,
        Duration::from_millis(100),
    )
    .await
    .unwrap();

    universe.assert_quit().await;
}