Improve

Author: iambriccardo

etl/tests/pipeline_with_partitioned_table.rs

@@ -67,9 +67,7 @@ async fn partitioned_table_copy_replicates_existing_data() {
 
     pipeline.start().await.unwrap();
 
-    tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified()).await;
 
     tokio::time::timeout(Duration::from_secs(3), pipeline.shutdown_and_wait())
         .await
@@ -156,9 +154,7 @@ async fn partitioned_table_copy_and_streams_new_data_from_new_partition() {
 
     pipeline.start().await.unwrap();
 
-    tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified()).await;
 
     let new_partition_name = format!("{}_{}", table_name.name, "p3");
     let new_partition_qualified_name = format!("{}.{}", table_name.schema, new_partition_name);
@@ -183,9 +179,7 @@ async fn partitioned_table_copy_and_streams_new_data_from_new_partition() {
     let inserts_notify = destination
         .wait_for_events_count(vec![(EventType::Insert, 1)])
         .await;
-    tokio::time::timeout(Duration::from_secs(3), inserts_notify.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), inserts_notify.notified()).await;
 
     tokio::time::timeout(Duration::from_secs(3), pipeline.shutdown_and_wait())
         .await
@@ -266,9 +260,7 @@ async fn partition_drop_does_not_emit_delete_or_truncate() {
     );
 
     pipeline.start().await.unwrap();
-    tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified()).await;
 
     let events_before = destination.get_events().await;
     let grouped_before = group_events_by_type_and_table_id(&events_before);
@@ -375,9 +367,7 @@ async fn partition_detach_with_explicit_publication_does_not_replicate_detached_
     );
 
     pipeline.start().await.unwrap();
-    tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified()).await;
 
     // Verify initial sync copied both rows.
     let table_rows = destination.get_table_rows().await;
@@ -424,9 +414,7 @@ async fn partition_detach_with_explicit_publication_does_not_replicate_detached_
     let inserts_notify = destination
         .wait_for_events_count(vec![(EventType::Insert, 1)])
         .await;
-    tokio::time::timeout(Duration::from_secs(3), inserts_notify.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), inserts_notify.notified()).await;
 
     tokio::time::timeout(Duration::from_secs(3), pipeline.shutdown_and_wait())
         .await
@@ -512,9 +500,7 @@ async fn partition_detach_with_all_tables_publication_does_not_replicate_detache
     );
 
     pipeline.start().await.unwrap();
-    tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified()).await;
 
     // Verify the initial state. The parent table is the only table tracked.
     let table_states_before = state_store.get_table_replication_states().await;
@@ -667,9 +653,7 @@ async fn partition_detach_with_all_tables_publication_does_replicate_detached_in
     );
 
     pipeline.start().await.unwrap();
-    tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified()).await;
 
     // Verify the initial state. The parent table is the only table tracked.
     let table_states_before = state_store.get_table_replication_states().await;
@@ -724,9 +708,7 @@ async fn partition_detach_with_all_tables_publication_does_replicate_detached_in
     pipeline.start().await.unwrap();
 
     // Wait for the detached partition to be synced.
-    tokio::time::timeout(Duration::from_secs(3), detached_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), detached_sync_done.notified()).await;
 
     tokio::time::timeout(Duration::from_secs(3), pipeline.shutdown_and_wait())
         .await
@@ -820,9 +802,7 @@ async fn partition_detach_with_schema_publication_does_not_replicate_detached_in
     );
 
     pipeline.start().await.unwrap();
-    tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified()).await;
 
     // Verify initial state.
     let table_states_before = state_store.get_table_replication_states().await;
@@ -886,9 +866,7 @@ async fn partition_detach_with_schema_publication_does_not_replicate_detached_in
     let inserts_notify = destination
         .wait_for_events_count(vec![(EventType::Insert, 1)])
         .await;
-    tokio::time::timeout(Duration::from_secs(3), inserts_notify.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), inserts_notify.notified()).await;
 
     tokio::time::timeout(Duration::from_secs(3), pipeline.shutdown_and_wait())
         .await
@@ -990,9 +968,7 @@ async fn partition_detach_with_schema_publication_does_replicate_detached_insert
     );
 
     pipeline.start().await.unwrap();
-    tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), parent_sync_done.notified()).await;
 
     // Verify initial state.
     let table_states_before = state_store.get_table_replication_states().await;
@@ -1047,9 +1023,7 @@ async fn partition_detach_with_schema_publication_does_replicate_detached_insert
     pipeline.start().await.unwrap();
 
     // Wait for the detached partition to be synced.
-    tokio::time::timeout(Duration::from_secs(3), detached_sync_done.notified())
-        .await
-        .unwrap();
+    let _ = tokio::time::timeout(Duration::from_secs(3), detached_sync_done.notified()).await;
 
     tokio::time::timeout(Duration::from_secs(3), pipeline.shutdown_and_wait())
         .await