|
7 | 7 | #![allow(clippy::single_range_in_vec_init)] |
8 | 8 |
|
9 | 9 | use crate::{ |
10 | | - InputBuffer, InputConsumer, ParseError, Parser, |
| 10 | + ControllerError, InputBuffer, InputConsumer, ParseError, Parser, |
11 | 11 | format::{Splitter, SpongeSplitter}, |
12 | 12 | test::{ |
13 | 13 | engine_test, input_pipeline, |
@@ -2175,7 +2175,13 @@ fn check_fault_tolerance( |
2175 | 2175 | .controller( |
2176 | 2176 | &circuit, |
2177 | 2177 | &config, |
2178 | | - Box::new(move |error, _| panic!("error: {error}")), |
| 2178 | + Box::new(move |error, _| { |
| 2179 | + if is_transient(&error) { |
| 2180 | + info!("ignored error: {error}") |
| 2181 | + } else { |
| 2182 | + panic!("error: {error}") |
| 2183 | + } |
| 2184 | + }), |
2179 | 2185 | ) |
2180 | 2186 | .unwrap(); |
2181 | 2187 | controller.start(); |
@@ -2946,18 +2952,18 @@ pub fn buffer(engine: &dyn PipelineTestEngine) { |
2946 | 2952 | .unwrap(); |
2947 | 2953 |
|
2948 | 2954 | // Once the topics are gone, endpoint threads may still report errors, which must not |
2949 | | - // fail the case. |
| 2955 | + // fail the case. Transient errors must not fail it either. |
2950 | 2956 | let running = Arc::new(AtomicBool::new(true)); |
2951 | 2957 | let running_clone = running.clone(); |
2952 | 2958 | let controller = engine |
2953 | 2959 | .controller( |
2954 | 2960 | &TestCircuit::identity(relation()), |
2955 | 2961 | &config, |
2956 | 2962 | Box::new(move |error, _| { |
2957 | | - if running_clone.load(Ordering::Acquire) { |
| 2963 | + if running_clone.load(Ordering::Acquire) && !is_transient(&error) { |
2958 | 2964 | panic!("buffer: error: {error}") |
2959 | 2965 | } else { |
2960 | | - info!("buffer: error after the topics are gone: {error}") |
| 2966 | + info!("buffer: ignored error: {error}") |
2961 | 2967 | } |
2962 | 2968 | }), |
2963 | 2969 | ) |
@@ -3012,6 +3018,16 @@ pub fn buffer(engine: &dyn PipelineTestEngine) { |
3012 | 3018 | running.store(false, Ordering::Release); |
3013 | 3019 | } |
3014 | 3020 |
|
| 3021 | +/// Returns true if `error` is a non-fatal transport error, such as `UnknownPartition` while |
| 3022 | +/// a new topic's metadata spreads across the brokers. The endpoint retries these itself. |
| 3023 | +fn is_transient(error: &ControllerError) -> bool { |
| 3024 | + matches!( |
| 3025 | + error, |
| 3026 | + ControllerError::InputTransportError { fatal: false, .. } |
| 3027 | + | ControllerError::OutputTransportError { fatal: false, .. } |
| 3028 | + ) |
| 3029 | +} |
| 3030 | + |
3015 | 3031 | /// Checks a pipeline that reads CSV `data` from an input topic and writes it in `format` |
3016 | 3032 | /// with `format_config` to an output topic, in messages of at most `message_max_bytes`. |
3017 | 3033 | fn check_end_to_end( |
@@ -3071,10 +3087,10 @@ fn check_end_to_end( |
3071 | 3087 | &TestCircuit::identity(relation()), |
3072 | 3088 | &config, |
3073 | 3089 | Box::new(move |error, _| { |
3074 | | - if running_clone.load(Ordering::Acquire) { |
| 3090 | + if running_clone.load(Ordering::Acquire) && !is_transient(&error) { |
3075 | 3091 | panic!("{name}: error: {error}") |
3076 | 3092 | } else { |
3077 | | - info!("{name}: error after the topics are gone: {error}") |
| 3093 | + info!("{name}: ignored error: {error}") |
3078 | 3094 | } |
3079 | 3095 | }), |
3080 | 3096 | ) |
|