Fix LocalStack S3 bucket setup in parquet test

Author: anovv

volga_stream/src/runtime/functions/source/parquet/integration_tests.rs

@@ -350,15 +350,16 @@ async fn parquet_sink_bounded_concurrency_backpressure() {
 #[ignore]
 async fn parquet_s3_roundtrip_localstack() {
     let docker = Cli::default();
-    let image = GenericImage::new("localstack/localstack", "latest")
+    let image = GenericImage::new("localstack/localstack", "3.0")
         .with_env_var("SERVICES", "s3")
         .with_env_var("DEFAULT_REGION", "us-east-1")
-        .with_env_var("DEFAULT_BUCKETS", "volga-test")
         .with_exposed_port(4566)
-        .with_wait_for(WaitFor::seconds(5));
-    let runnable = RunnableImage::from(image);
+        .with_wait_for(WaitFor::seconds(3));
+    let runnable = RunnableImage::from(image).with_mapped_port((4566, 4566));
     let container = docker.run(runnable);
     let endpoint = format!("http://127.0.0.1:{}", container.get_host_port_ipv4(4566));
+    wait_for_localstack_ready("127.0.0.1:4566").await;
+    create_localstack_bucket(&endpoint, "volga-test").await;
 
     let schema = test_schema();
     let opts = HashMap::from([
@@ -376,3 +377,28 @@ async fn parquet_s3_roundtrip_localstack() {
     )
     .await;
 }
+
+async fn wait_for_localstack_ready(addr: &str) {
+    for _ in 0..40 {
+        if tokio::net::TcpStream::connect(addr).await.is_ok() {
+            return;
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
+    }
+    panic!("LocalStack did not become ready at {}", addr);
+}
+
+async fn create_localstack_bucket(endpoint: &str, bucket: &str) {
+    let url = format!("{}/{}", endpoint, bucket);
+    let client = reqwest::Client::new();
+    for _ in 0..10 {
+        match client.put(&url).send().await {
+            Ok(resp) if resp.status().is_success() || resp.status().as_u16() == 409 => {
+                return;
+            }
+            _ => {}
+        }
+        tokio::time::sleep(std::time::Duration::from_millis(250)).await;
+    }
+    panic!("Failed to create LocalStack bucket at {}", url);
+}