Merge branch 'main' into dev/sapatr/codesimplification1

Author: sapatrjv

rust/otap-dataflow/.config/nextest.toml

@@ -4,7 +4,7 @@ retries = 2
 
 [profile.ci]
 slow-timeout = { period = "60s", terminate-after = 2 }
-max-fail = 10
+fail-fast = { max-fail = 10 }
 retries = 2
 
 # JUnit output for flaky test tracking.

rust/otap-dataflow/crates/core-nodes/src/exporters/otap_exporter/mod.rs

@@ -852,9 +852,7 @@ mod tests {
         );
     }
 
-    // Skipping on Windows due to flakiness: https://github.com/open-telemetry/otel-arrow/issues/1614
     #[test]
-    #[cfg_attr(windows, ignore = "Skipping on Windows due to flakiness")]
     fn test_receiver_not_ready_on_start() {
         let grpc_addr = "127.0.0.1";
         let grpc_port = portpicker::pick_unused_port().expect("No free ports");

rust/otap-dataflow/crates/core-nodes/src/exporters/otlp_grpc_exporter/mod.rs

@@ -862,13 +862,25 @@ mod tests {
 
     use otap_df_config::node::NodeUserConfig;
 
-    #[cfg(not(windows))]
+    use otap_df_config::transport_headers::{TransportHeader, TransportHeaders};
     use otap_df_config::transport_headers_policy::PropagationSelectorType;
+    use otap_df_config::transport_headers_policy::{
+        HeaderPropagationPolicy, PropagationAction, PropagationDefault, PropagationMatch,
+        PropagationOverride, PropagationSelector,
+    };
     use otap_df_engine::Interests;
     use otap_df_engine::context::ControllerContext;
     use otap_df_engine::control::PipelineCompletionMsg;
+    use otap_df_engine::control::{
+        Controllable, PipelineCompletionMsgSender, RuntimeCtrlMsgSender,
+        pipeline_completion_msg_channel, runtime_ctrl_msg_channel,
+    };
     use otap_df_engine::error::Error;
     use otap_df_engine::exporter::ExporterWrapper;
+    use otap_df_engine::local::message::{LocalReceiver, LocalSender};
+    use otap_df_engine::message::{Receiver, Sender};
+    use otap_df_engine::node::NodeWithPDataReceiver;
+    use otap_df_engine::testing::create_not_send_channel;
     use otap_df_engine::testing::{
         exporter::{TestContext, TestRuntime},
         test_node,
@@ -883,7 +895,9 @@ mod tests {
     use otap_df_pdata::proto::opentelemetry::collector::metrics::v1::metrics_service_server::MetricsServiceServer;
     use otap_df_pdata::proto::opentelemetry::collector::trace::v1::ExportTraceServiceRequest;
     use otap_df_pdata::proto::opentelemetry::collector::trace::v1::trace_service_server::TraceServiceServer;
+    use otap_df_telemetry::metrics::MetricSetSnapshot;
     use otap_df_telemetry::registry::TelemetryRegistryHandle;
+    use otap_df_telemetry::reporter::MetricsReporter;
     use prost::Message;
     use std::net::SocketAddr;
     use std::pin::Pin;
@@ -893,25 +907,6 @@ mod tests {
     use tokio::time::{Duration, timeout};
     use tonic::codegen::tokio_stream::wrappers::TcpListenerStream;
     use tonic::transport::Server;
-    // Imports only used by tests that are skipped on Windows
-    #[cfg(not(windows))]
-    use {
-        otap_df_config::transport_headers::{TransportHeader, TransportHeaders},
-        otap_df_config::transport_headers_policy::{
-            HeaderPropagationPolicy, PropagationAction, PropagationDefault, PropagationMatch,
-            PropagationOverride, PropagationSelector,
-        },
-        otap_df_engine::control::{
-            Controllable, PipelineCompletionMsgSender, RuntimeCtrlMsgSender,
-            pipeline_completion_msg_channel, runtime_ctrl_msg_channel,
-        },
-        otap_df_engine::local::message::{LocalReceiver, LocalSender},
-        otap_df_engine::message::{Receiver, Sender},
-        otap_df_engine::node::NodeWithPDataReceiver,
-        otap_df_engine::testing::create_not_send_channel,
-        otap_df_telemetry::metrics::MetricSetSnapshot,
-        otap_df_telemetry::reporter::MetricsReporter,
-    };
 
     /// Helper function to wait for and validate an Ack or Nack message with the expected node_id
     async fn wait_for_ack_or_nack(
@@ -1153,8 +1148,6 @@ mod tests {
         _ = shutdown_sender.send("Shutdown");
     }
 
-    // Skipping on Windows due to flakiness: https://github.com/open-telemetry/otel-arrow/issues/1611
-    #[cfg(not(windows))]
     #[test]
     fn test_receiver_not_ready_on_start_and_reconnect() {
         // the purpose of this test is to that the exporter behaves as expected in the face of
@@ -1181,6 +1174,7 @@ mod tests {
                 config: Config {
                     grpc: GrpcClientSettings {
                         grpc_endpoint: grpc_endpoint.clone(),
+                        connect_timeout: Duration::from_millis(500),
                         ..Default::default()
                     },
                     max_in_flight: 32,
@@ -1458,7 +1452,6 @@ mod tests {
     // ---- build_grpc_metadata unit tests ----------------------------------------
 
     /// Helper: Creates an [`EffectHandler`] with an optional propagation policy set.
-    #[cfg(not(windows))]
     fn make_effect_handler_with_policy(
         policy: Option<HeaderPropagationPolicy>,
     ) -> EffectHandler<OtapPdata> {
@@ -1470,7 +1463,6 @@ mod tests {
     }
 
     /// Helper: Creates a [`Context`] that carries the given transport headers.
-    #[cfg(not(windows))]
     fn context_with_headers(headers: TransportHeaders) -> Context {
         let pdata = OtapPdata::new_default(OtlpProtoBytes::ExportLogsRequest(Bytes::new()).into())
             .with_transport_headers(headers);
@@ -1479,15 +1471,13 @@ mod tests {
     }
 
     /// Helper: Creates a [`Context`] without any transport headers.
-    #[cfg(not(windows))]
     fn context_without_headers() -> Context {
         let pdata = OtapPdata::new_default(OtlpProtoBytes::ExportLogsRequest(Bytes::new()).into());
         let (context, _) = pdata.into_parts();
         context
     }
 
     /// Helper: Propagation policy that propagates all captured headers.
-    #[cfg(not(windows))]
     fn propagate_all_policy() -> HeaderPropagationPolicy {
         HeaderPropagationPolicy::new(
             PropagationDefault {
@@ -1501,7 +1491,6 @@ mod tests {
         )
     }
 
-    #[cfg(not(windows))]
     #[test]
     fn test_build_grpc_metadata_returns_none_without_policy() {
         let handler = make_effect_handler_with_policy(None);
@@ -1513,7 +1502,6 @@ mod tests {
         assert!(result.is_none(), "should return None when no policy is set");
     }
 
-    #[cfg(not(windows))]
     #[test]
     fn test_build_grpc_metadata_returns_none_without_headers() {
         let handler = make_effect_handler_with_policy(Some(propagate_all_policy()));
@@ -1526,7 +1514,6 @@ mod tests {
         );
     }
 
-    #[cfg(not(windows))]
     #[test]
     fn test_build_grpc_metadata_propagates_text_headers() {
         let handler = make_effect_handler_with_policy(Some(propagate_all_policy()));
@@ -1558,7 +1545,6 @@ mod tests {
         assert_eq!(request_id.to_str().unwrap(), "req-xyz-789");
     }
 
-    #[cfg(not(windows))]
     #[test]
     fn test_build_grpc_metadata_drops_filtered_headers() {
         let policy = HeaderPropagationPolicy::new(
@@ -1602,7 +1588,6 @@ mod tests {
         );
     }
 
-    #[cfg(not(windows))]
     #[test]
     fn test_build_grpc_metadata_propagates_binary_headers() {
         let handler = make_effect_handler_with_policy(Some(propagate_all_policy()));
@@ -1629,7 +1614,6 @@ mod tests {
         );
     }
 
-    #[cfg(not(windows))]
     #[test]
     fn test_build_grpc_metadata_appends_bin_suffix_for_binary_headers() {
         let handler = make_effect_handler_with_policy(Some(propagate_all_policy()));
@@ -1653,7 +1637,6 @@ mod tests {
         assert_eq!(bin_val.to_bytes().unwrap(), binary_value.as_slice());
     }
 
-    #[cfg(not(windows))]
     #[test]
     fn test_build_grpc_metadata_preserves_duplicate_headers() {
         let handler = make_effect_handler_with_policy(Some(propagate_all_policy()));
@@ -1691,7 +1674,6 @@ mod tests {
         );
     }
 
-    #[cfg(not(windows))]
     #[test]
     fn test_build_grpc_metadata_returns_none_when_all_dropped() {
         // Policy that drops everything (selector = None means no headers are selected).

rust/otap-dataflow/crates/core-nodes/src/exporters/parquet_exporter/mod.rs

@@ -250,6 +250,19 @@ impl Exporter<OtapPdata> for ParquetExporter {
                     deadline,
                     reason: _,
                 }) => {
+                    // If the deadline has already passed, return immediately.
+                    // `Delay::new(Duration::ZERO)` is not guaranteed to resolve
+                    // on the first poll on all platforms (Windows timer
+                    // granularity is ~15 ms), so an explicit check avoids a
+                    // race between the timeout and the flush future.
+                    if deadline.checked_duration_since(Instant::now()).is_none() {
+                        let _ = telemetry_cancel_handle.cancel().await;
+                        return Err(Error::IoError {
+                            node: exporter_id.clone(),
+                            error: std::io::Error::from(ErrorKind::TimedOut),
+                        });
+                    }
+
                     let mut timeout = Delay::new(deadline.duration_since(Instant::now())).fuse();
                     let flush_all = writer.flush_all().fuse();
                     pin_mut!(flush_all);
@@ -503,10 +516,6 @@ mod test {
     }
 
     #[test]
-    #[cfg_attr(
-        target_os = "windows",
-        ignore = "Skipping on Windows due to timing flakiness"
-    )]
     fn test_adaptive_schema_dict_upgrade_write() {
         let test_runtime = TestRuntime::<OtapPdata>::new();
         let temp_dir = tempfile::tempdir().unwrap();
@@ -895,19 +904,15 @@ mod test {
             });
     }
 
-    // Skipping on Windows and macOS due to flakiness: https://github.com/open-telemetry/otel-arrow/issues/1614
     #[test]
-    #[cfg_attr(
-        any(target_os = "windows", target_os = "macos"),
-        ignore = "Skipping on Windows and macOS due to flakiness"
-    )]
     fn test_shutdown_timeout() {
         let test_runtime = TestRuntime::<OtapPdata>::new();
         let temp_dir = tempfile::tempdir().unwrap();
         let base_dir: String = temp_dir.path().to_str().unwrap().into();
+        let base_dir_url = base_dir.replace('\\', "/");
         let exporter = ParquetExporter::new(config::Config {
             storage: object_store::StorageType::File {
-                base_uri: format!("testdelayed://{base_dir}?delay=500ms"),
+                base_uri: format!("testdelayed:///{base_dir_url}?delay=500ms"),
             },
             partitioning_strategies: None,
             writer_options: Some(WriterOptions {
@@ -1030,10 +1035,10 @@ mod test {
                 );
             }
 
-            // shutdown faster than it could possibly flush
+            // Make timeout deterministic: deadline is already due when shutdown is handled.
             _ = ctrl_sender
                 .send(NodeControlMsg::Shutdown {
-                    deadline: Instant::now().add(Duration::from_secs(1)),
+                    deadline: Instant::now(),
                     reason: "shutting down".into(),
                 })
                 .await;

rust/otap-dataflow/crates/core-nodes/src/receivers/otlp_receiver/mod.rs

@@ -1758,7 +1758,7 @@ mod tests {
                     .await
                     .expect("Failed to send Shutdown");
 
-                timeout(Duration::from_secs(1), async {
+                timeout(Duration::from_secs(5), async {
                     loop {
                         if LogsServiceClient::connect(grpc_endpoint.clone())
                             .await
@@ -1866,7 +1866,6 @@ mod tests {
         }
     }
 
-    #[cfg_attr(any(windows), ignore = "Skipping on Windows due to flakiness")]
     #[test]
     fn test_otlp_receiver_ack() {
         let test_runtime = TestRuntime::new();

rust/otap-dataflow/crates/otap/src/object_store.rs

@@ -253,6 +253,14 @@ mod test {
 
         let path = url.path().to_string();
 
+        // On Windows, url.path() returns "/C:/..." for file paths; strip the leading slash
+        // so that LocalFileSystem receives a valid Windows path.
+        #[cfg(windows)]
+        let path = path
+            .strip_prefix('/')
+            .map(|s| s.to_string())
+            .unwrap_or(path);
+
         let delay = url
             .query_pairs()
             .find(|(k, _)| k == "delay")
@@ -342,23 +350,20 @@ mod test {
         }
     }
 
-    // Skipping on Windows: https://github.com/open-telemetry/otel-arrow/issues/1614
     #[test]
-    #[cfg(not(windows))]
     fn test_get_testdelayed_file_storage() {
-        let storage = StorageType::File {
-            base_uri: "testdelayed:///tmp".to_string(),
-        };
+        let tmp = tempfile::tempdir().unwrap();
+        let path = tmp.path().to_str().unwrap().replace('\\', "/");
+        let base_uri = format!("testdelayed:///{path}");
+        let storage = StorageType::File { base_uri };
         assert!(from_storage_type(&storage).is_ok());
     }
 
-    // Skipping on Windows: https://github.com/open-telemetry/otel-arrow/issues/1614
     #[test]
-    #[cfg(not(windows))]
     fn test_get_file_storage() {
-        let storage = StorageType::File {
-            base_uri: "/tmp".to_string(),
-        };
+        let tmp = tempfile::tempdir().unwrap();
+        let base_uri = tmp.path().to_str().unwrap().to_string();
+        let storage = StorageType::File { base_uri };
         assert!(from_storage_type(&storage).is_ok());
     }
 

rust/otap-dataflow/crates/otap/src/otap_grpc/client_settings.rs

@@ -602,10 +602,6 @@ mod tests {
     }
 
     #[tokio::test]
-    #[cfg_attr(
-        any(target_os = "windows", target_os = "macos"),
-        ignore = "Skipping on Windows and macOS due to flakiness"
-    )]
     async fn build_endpoint_with_tls_errors_when_ca_file_missing() {
         crate::crypto::ensure_crypto_provider();
         let settings = GrpcClientSettings {
@@ -619,9 +615,12 @@ mod tests {
         };
 
         let err = settings.build_endpoint_with_tls().await.unwrap_err();
+        let err_msg = err.to_string().to_lowercase();
         assert!(
-            err.to_string().to_lowercase().contains("no such")
-                || err.to_string().to_lowercase().contains("not found")
+            err_msg.contains("no such")
+                || err_msg.contains("not found")
+                || err_msg.contains("cannot find"),
+            "unexpected error message: {err_msg}"
         );
     }