Function test_no_block

src/environmentd/tests/sql.rs:112–209

()

#[mz_ore::test(tokio::test(flavor = "multi_thread", worker_threads = 1))]
#[allow(clippy::disallowed_methods)]
async fn test_no_block() {
    // We manually time out the test because it's better than relying on CI to time out, because
    // an actual failure (as opposed to a CI timeout) causes `services.log` to be uploaded.

    // Allow the use of banned rdkafka methods, because we are just in tests.
    #[allow(clippy::disallowed_methods)]
    let test_case = async move {
        println!("test_no_block: starting server");
        let server = test_util::TestHarness::default().start().await;
        server
            .enable_feature_flags(&["enable_connection_validation_syntax"])
            .await;

        println!("test_no_block: starting mock HTTP server");
        let mut schema_registry_server = MockHttpServer::new().await;

        println!("test_no_block: connecting to server");
        let client = server.connect().await.unwrap();

        let slow_task = task::spawn(|| "slow_client", async move {
            println!("test_no_block: in thread; executing create source");
            let result = client
                .batch_execute(&format!(
                    "CREATE CONNECTION IF NOT EXISTS csr_conn TO CONFLUENT SCHEMA REGISTRY (URL 'http://{}') WITH (VALIDATE = false);",
                    schema_registry_server.addr,
                ))
                .await;
            println!("test_no_block: in thread; create CSR conn done");
            result.unwrap();

            let admin: AdminClient<_> = ClientConfig::new()
                .set("bootstrap.servers", &*KAFKA_ADDRS)
                .create()
                .expect("Admin client creation failed");

            let new_topic = NewTopic::new("foo", 1, TopicReplication::Fixed(1));
            let topic_results = admin
                .create_topics([&new_topic], &AdminOptions::new())
                .await
                .expect("topic creation failed");
            match topic_results[0] {
                Ok(_) | Err((_, RDKafkaErrorCode::TopicAlreadyExists)) => {}
                Err((ref err, _)) => panic!("failed to ensure topic: {err}"),
            }

            let result = client
                    .batch_execute(&format!(
                        "CREATE CONNECTION kafka_conn TO KAFKA (BROKER '{}', SECURITY PROTOCOL PLAINTEXT) WITH (VALIDATE = false)",
                        &*KAFKA_ADDRS,
                    ))
                    .await;
            println!("test_no_block: in thread; create Kafka conn done");
            result.unwrap();

            let result = client
                .batch_execute(
                    "CREATE SOURCE foo \
                        FROM KAFKA CONNECTION kafka_conn (TOPIC 'foo') \