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/clickhouse
    
    the code of the file until where you have to start completion: use std::{
    convert::Infallible,
    net::SocketAddr,
    sync::{
        atomic::{AtomicBool, Ordering},
        Arc,
    },
};

use futures::{
    future::{ok, ready},
    stream,
};
use http::StatusCode;
use serde::Deserialize;
use serde_json::Value;
use tokio::time::{timeout, Duration};
use vector_lib::event::{BatchNotifier, BatchStatus, BatchStatusReceiver, Event, LogEvent};
use vector_lib::lookup::PathPrefix;
use warp::Filter;

use super::*;
use crate::{
    codecs::{TimestampFormat, Transformer},
    config::{log_schema, SinkConfig, SinkContext},
    
async fn skip_unknown_fields() {
    trace_init();

    let table = gen_table();
    let host = clickhouse_address();

    let mut batch = BatchConfig::default();
    batch.max_events = Some(1);

    let config = ClickhouseConfig {
        endpoint: host.parse().unwrap(),
        table: table.clone().try_into().unwrap(),
        skip_unknown_fields: true,
        compression: Compression::None,
        batch,
        request: TowerRequestConfig {
            retry_attempts: 1,
            ..Default::default()
        },
        ..Default::default()
    };

    let client = ClickhouseClient::new(host);
    client
        .create_table(&table, "host String, timestamp String, message String")
        .await;

    let (sink, _hc) = config.build(SinkContext::default()).await.unwrap();

    let (mut input_event, mut receiver) = make_event();
    input_event.as_mut_log().insert("unknown", "mysteries");

    run_and_assert_sink_compliance(sink, stream::once(ready(input_event.clone())), &SINK_TAGS)
        .await;

    let output = client.select_all(&table).await;
    assert_eq!(1, output.rows);

    input_event.as_mut_log().remove("unknown");
    let expected = serde_json::to_value(input_event.into_log()).unwrap();
    assert_eq!(expected, output.data[0]);

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