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: integration_tests.rs
    path of file: ./repos/vector/src/sinks/elasticsearch
    
    the code of the file until where you have to start completion: use std::{fs::File, io::Read};

use aws_smithy_types::body::SdkBody;
use bytes::Bytes;
use chrono::Utc;
use futures::StreamExt;
use futures::{future::ready, stream};
use http::{Request, StatusCode};
use serde_json::{json, Value};
use vector_lib::{
    config::{init_telemetry, log_schema, Tags, Telemetry},
    event::{BatchNotifier, BatchStatus, Event, LogEvent},
};

use super::{config::DATA_STREAM_TIMESTAMP_KEY, *};
use crate::{
    aws::{ImdsAuthentication, RegionOrEndpoint},
    config::{ProxyConfig, SinkConfig, SinkContext},
    http::HttpClient,
    sinks::{
        util::{auth::Auth, BatchConfig, Compression, SinkBatchSettings},
        HealthcheckError,
    },
    test_util::{
        components::{
            run_and_assert_data_volume_sink_compliance, run_and_assert_sink_compliance,
            run_and_assert_sink_error, COMPONENT_ERROR_TAGS, DATA_VOLUME_SINK_TAGS, HTTP_SINK_TAGS,
        },
        random_events_with_stream, random_string, trace_init,
    },
    tls::{self, TlsConfig},
};

fn aws_server() -> String {
    std::env::var("E
    
async fn run_insert_tests_with_multiple_endpoints(config: &ElasticsearchConfig) {
    let cx = SinkContext::default();
    let commons = ElasticsearchCommon::parse_many(config, cx.proxy())
        .await
        .expect("Config error");
    let index = match config.mode {
        // Data stream mode uses an index name generated from the event.
        ElasticsearchMode::DataStream => format!(
            "{}",
            Utc::now().format(".ds-logs-generic-default-%Y.%m.%d-000001")
        ),
        ElasticsearchMode::Bulk => config.bulk.index.to_string(),
    };

    let (sink, healthcheck) = config
        .build(cx.clone())
        .await
        .expect("Building config failed");

    healthcheck.await.expect("Health check failed");

    let (batch, mut receiver) = BatchNotifier::new_with_receiver();
    let (input, events) = random_events_with_stream(100, 100, Some(batch));
    run_and_assert_sink_compliance(sink, events, &HTTP_SINK_TAGS).await;

    assert_eq!(receiver.try_recv(), Ok(BatchStatus::Delivered));

    let base_urls = commons
        .iter()
        .map(|common| common.base_url.clone())
        .collect::<Vec<_>>();

    // make sure writes are all visible
    for common in commons {
        _ = flush(common).await;
    }

    let client = create_http_client();
    let mut total = 0;
    for base_url in base_urls {
        if let Ok(response) = client
            .get(&format!("{}/{}/_search", base_url, index))
            .basic_auth("elastic", Some("vector"))
            .json(&json!({
                "query": { "query_string": { "query": "*" } }
            }))
            .send()
            .await
        {
            let response = response.json::<Value>().await.unwrap();

            let endpoint_total = response["hits"]["total"]["value"]
                .as_u64()
                .or_else(|| response["hits"]["total"].as_u64())
                .expect(
                    "Elasticsearch response does not include hits->total nor hits->total->value",
                );

            assert!(
                input.len() as u64 > endpoint_total,
                "One of the endpoints received all of the events."
            );

            total += endpoint_total;
        }
    }

    assert_eq!(input.len() as u64, total);
}