5725710489

pending completion

Build # 5725710489

Build Type

push

github

Committed by

web-flow

Commit Message

chore: Add `SourceFactory::get_output_port_name` to simplify ui graph generation (#1812)

Run Details

140 of 140 new or added lines in 13 files covered. (100.0%)

45519 of 60083 relevant lines covered (75.76%)

39458.21 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.15

/dozer-core/src/tests/dag_base_errors.rs

use crate::channels::{ProcessorChannelForwarder, SourceChannelForwarder};
use crate::executor::{DagExecutor, ExecutorOptions};
use crate::executor_operation::ProcessorOperation;
use crate::node::{
    OutputPortDef, OutputPortType, PortHandle, Processor, ProcessorFactory, Sink, SinkFactory,
    Source, SourceFactory,
};
use crate::tests::dag_base_run::NoopProcessorFactory;
use crate::tests::sinks::{CountingSinkFactory, COUNTING_SINK_INPUT_PORT};
use crate::tests::sources::{GeneratorSourceFactory, GENERATOR_SOURCE_OUTPUT_PORT};
use crate::{Dag, Endpoint, DEFAULT_PORT_HANDLE};
use dozer_types::epoch::Epoch;
use dozer_types::errors::internal::BoxedError;
use dozer_types::ingestion_types::IngestionMessage;
use dozer_types::node::NodeHandle;
use dozer_types::types::{
    Field, FieldDefinition, FieldType, Operation, Record, Schema, SourceDefinition,
};

use std::collections::HashMap;
use std::panic;

use std::sync::atomic::AtomicBool;
use std::sync::Arc;

use crate::tests::app::NoneContext;

// Test when error is generated by a processor

#[derive(Debug)]
struct ErrorProcessorFactory {
    err_on: u64,
    panic: bool,
}

impl ProcessorFactory<NoneContext> for ErrorProcessorFactory {
    fn type_name(&self) -> String {
        "Error".to_owned()
    }

    fn get_output_schema(
        &self,
        _output_port: &PortHandle,
        input_schemas: &HashMap<PortHandle, (Schema, NoneContext)>,
    ) -> Result<(Schema, NoneContext), BoxedError> {
        Ok(input_schemas.get(&DEFAULT_PORT_HANDLE).unwrap().clone())
    }

    fn get_input_ports(&self) -> Vec<PortHandle> {
        vec![DEFAULT_PORT_HANDLE]
    }

    fn get_output_ports(&self) -> Vec<OutputPortDef> {
        vec![OutputPortDef::new(
            DEFAULT_PORT_HANDLE,
            OutputPortType::Stateless,
        )]
    }

    fn build(
        &self,
        _input_schemas: HashMap<PortHandle, Schema>,
        _output_schemas: HashMap<PortHandle, Schema>,
    ) -> Result<Box<dyn Processor>, BoxedError> {
        Ok(Box::new(ErrorProcessor {
            err_on: self.err_on,
            count: 0,
            panic: self.panic,
        }))
    }

    fn id(&self) -> String {
        "Error".to_owned()
    }
}

#[derive(Debug)]
struct ErrorProcessor {
    err_on: u64,
    count: u64,
    panic: bool,
}

impl Processor for ErrorProcessor {
    fn commit(&self, _epoch: &Epoch) -> Result<(), BoxedError> {
        Ok(())
    }

    fn process(
        &mut self,
        _from_port: PortHandle,
        op: ProcessorOperation,
        fw: &mut dyn ProcessorChannelForwarder,
    ) -> Result<(), BoxedError> {
        self.count += 1;
        if self.count == self.err_on {
            if self.panic {
                panic!("Generated error");
            } else {
                return Err("Uknown".to_string().into());
            }
        }

        fw.send(op, DEFAULT_PORT_HANDLE);
        Ok(())
    }
}

#[test]
#[should_panic]
fn test_run_dag_proc_err_panic() {
    let count: u64 = 1_000_000;

    let mut dag = Dag::new();
    let latch = Arc::new(AtomicBool::new(true));

    let source_handle = NodeHandle::new(None, 1.to_string());
    let proc_handle = NodeHandle::new(Some(1), 1.to_string());
    let sink_handle = NodeHandle::new(Some(1), 2.to_string());

    dag.add_source(
        source_handle.clone(),
        Box::new(GeneratorSourceFactory::new(count, latch.clone(), false)),
    );
    dag.add_processor(
        proc_handle.clone(),
        Box::new(ErrorProcessorFactory {
            err_on: 800_000,
            panic: true,
        }),
    );
    dag.add_sink(
        sink_handle.clone(),
        Box::new(CountingSinkFactory::new(count, latch)),
    );

    dag.connect(
        Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),
        Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),
    )
    .unwrap();

    dag.connect(
        Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),
        Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),
    )
    .unwrap();

    DagExecutor::new(dag, ExecutorOptions::default())
        .unwrap()
        .start(Arc::new(AtomicBool::new(true)))
        .unwrap()
        .join()
        .unwrap();
}

#[test]
#[should_panic]
fn test_run_dag_proc_err_2() {
    let count: u64 = 1_000_000;

    let mut dag = Dag::new();
    let latch = Arc::new(AtomicBool::new(true));

    let source_handle = NodeHandle::new(None, 1.to_string());
    let proc_handle = NodeHandle::new(Some(1), 1.to_string());
    let proc_err_handle = NodeHandle::new(Some(1), 2.to_string());
    let sink_handle = NodeHandle::new(Some(1), 3.to_string());

    dag.add_source(
        source_handle.clone(),
        Box::new(GeneratorSourceFactory::new(count, latch.clone(), false)),
    );
    dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));

    dag.add_processor(
        proc_err_handle.clone(),
        Box::new(ErrorProcessorFactory {
            err_on: 800_000,
            panic: false,
        }),
    );

    dag.add_sink(
        sink_handle.clone(),
        Box::new(CountingSinkFactory::new(count, latch)),
    );

    dag.connect(
        Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),
        Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),
    )
    .unwrap();

    dag.connect(
        Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),
        Endpoint::new(proc_err_handle.clone(), DEFAULT_PORT_HANDLE),
    )
    .unwrap();

    dag.connect(
        Endpoint::new(proc_err_handle, DEFAULT_PORT_HANDLE),
        Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),
    )
    .unwrap();

    DagExecutor::new(dag, ExecutorOptions::default())
        .unwrap()
        .start(Arc::new(AtomicBool::new(true)))
        .unwrap()
        .join()
        .unwrap();
}

#[test]
#[should_panic]
fn test_run_dag_proc_err_3() {
    let count: u64 = 1_000_000;

    let mut dag = Dag::new();
    let latch = Arc::new(AtomicBool::new(true));

    let source_handle = NodeHandle::new(None, 1.to_string());
    let proc_handle = NodeHandle::new(Some(1), 1.to_string());
    let proc_err_handle = NodeHandle::new(Some(1), 2.to_string());
    let sink_handle = NodeHandle::new(Some(1), 3.to_string());

    dag.add_source(
        source_handle.clone(),
        Box::new(GeneratorSourceFactory::new(count, latch.clone(), false)),
    );

    dag.add_processor(
        proc_err_handle.clone(),
        Box::new(ErrorProcessorFactory {
            err_on: 800_000,
            panic: false,
        }),
    );

    dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));

    dag.add_sink(
        sink_handle.clone(),
        Box::new(CountingSinkFactory::new(count, latch)),
    );

    dag.connect(
        Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),
        Endpoint::new(proc_err_handle.clone(), DEFAULT_PORT_HANDLE),
    )
    .unwrap();

    dag.connect(
        Endpoint::new(proc_err_handle, DEFAULT_PORT_HANDLE),
        Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),
    )
    .unwrap();

    dag.connect(
        Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),
        Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),
    )
    .unwrap();

    DagExecutor::new(dag, ExecutorOptions::default())
        .unwrap()
        .start(Arc::new(AtomicBool::new(true)))
        .unwrap()
        .join()
        .unwrap();
}

// Test when error is generated by a source

#[derive(Debug)]
pub(crate) struct ErrGeneratorSourceFactory {
    count: u64,
    err_at: u64,
}

impl ErrGeneratorSourceFactory {
    pub fn new(count: u64, err_at: u64) -> Self {
        Self { count, err_at }
    }
}

impl SourceFactory<NoneContext> for ErrGeneratorSourceFactory {
    fn get_output_schema(&self, _port: &PortHandle) -> Result<(Schema, NoneContext), BoxedError> {
        Ok((
            Schema::default()
                .field(
                    FieldDefinition::new(
                        "id".to_string(),
                        FieldType::String,
                        false,
                        SourceDefinition::Dynamic,
                    ),
                    true,
                )
                .field(
                    FieldDefinition::new(
                        "value".to_string(),
                        FieldType::String,
                        false,
                        SourceDefinition::Dynamic,
                    ),
                    false,
                )
                .clone(),
            NoneContext {},
        ))
    }

    fn get_output_port_name(&self, _port: &PortHandle) -> String {
        "error".to_string()
    }

    fn get_output_ports(&self) -> Vec<OutputPortDef> {
        vec![OutputPortDef::new(
            GENERATOR_SOURCE_OUTPUT_PORT,
            OutputPortType::Stateless,
        )]
    }

    fn build(
        &self,
        _output_schemas: HashMap<PortHandle, Schema>,
    ) -> Result<Box<dyn Source>, BoxedError> {
        Ok(Box::new(ErrGeneratorSource {
            count: self.count,
            err_at: self.err_at,
        }))
    }
}

#[derive(Debug)]
pub(crate) struct ErrGeneratorSource {
    count: u64,
    err_at: u64,
}

impl Source for ErrGeneratorSource {
    fn can_start_from(&self, _last_checkpoint: (u64, u64)) -> Result<bool, BoxedError> {
        Ok(false)
    }

    fn start(
        &self,
        fw: &mut dyn SourceChannelForwarder,
        _checkpoint: Option<(u64, u64)>,
    ) -> Result<(), BoxedError> {
        for n in 1..(self.count + 1) {
            if n == self.err_at {
                return Err("Generated Error".to_string().into());
            }

            fw.send(
                IngestionMessage::new_op(
                    n,
                    0,
                    0,
                    Operation::Insert {
                        new: Record::new(vec![
                            Field::String(format!("key_{n}")),
                            Field::String(format!("value_{n}")),
                        ]),
                    },
                ),
                GENERATOR_SOURCE_OUTPUT_PORT,
            )?;
        }
        Ok(())
    }
}

#[test]
fn test_run_dag_src_err() {
    let count: u64 = 1_000_000;

    let mut dag = Dag::new();
    let latch = Arc::new(AtomicBool::new(true));

    let source_handle = NodeHandle::new(None, 1.to_string());
    let proc_handle = NodeHandle::new(Some(1), 1.to_string());
    let sink_handle = NodeHandle::new(Some(1), 3.to_string());

    dag.add_source(
        source_handle.clone(),
        Box::new(ErrGeneratorSourceFactory::new(count, 200_000)),
    );
    dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));
    dag.add_sink(
        sink_handle.clone(),
        Box::new(CountingSinkFactory::new(count, latch)),
    );

    dag.connect(
        Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),
        Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),
    )
    .unwrap();

    dag.connect(
        Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),
        Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),
    )
    .unwrap();

    let _join_handle = DagExecutor::new(dag, ExecutorOptions::default())
        .unwrap()
        .start(Arc::new(AtomicBool::new(true)))
        .unwrap();
    // join_handle.join().unwrap();
}

#[derive(Debug)]
pub(crate) struct ErrSinkFactory {
    err_at: u64,
    panic: bool,
}

impl ErrSinkFactory {
    pub fn new(err_at: u64, panic: bool) -> Self {
        Self { err_at, panic }
    }
}

impl SinkFactory<NoneContext> for ErrSinkFactory {
    fn get_input_ports(&self) -> Vec<PortHandle> {
        vec![COUNTING_SINK_INPUT_PORT]
    }

    fn prepare(
        &self,
        _input_schemas: HashMap<PortHandle, (Schema, NoneContext)>,
    ) -> Result<(), BoxedError> {
        Ok(())
    }

    fn build(
        &self,
        _input_schemas: HashMap<PortHandle, Schema>,
    ) -> Result<Box<dyn Sink>, BoxedError> {
        Ok(Box::new(ErrSink {
            err_at: self.err_at,
            current: 0,
            panic: self.panic,
        }))
    }
}

#[derive(Debug)]
pub(crate) struct ErrSink {
    err_at: u64,
    current: u64,
    panic: bool,
}
impl Sink for ErrSink {
    fn commit(&mut self, _epoch_details: &Epoch) -> Result<(), BoxedError> {
        Ok(())
    }

    fn process(
        &mut self,
        _from_port: PortHandle,
        _op: ProcessorOperation,
    ) -> Result<(), BoxedError> {
        self.current += 1;
        if self.current == self.err_at {
            if self.panic {
                panic!("Generated error");
            } else {
                return Err("Generated error".to_string().into());
            }
        }
        Ok(())
    }

    fn on_source_snapshotting_done(&mut self, _connection_name: String) -> Result<(), BoxedError> {
        Ok(())
    }
}

#[test]
#[should_panic]
fn test_run_dag_sink_err() {
    let count: u64 = 1_000_000;

    let mut dag = Dag::new();
    let latch = Arc::new(AtomicBool::new(true));

    let source_handle = NodeHandle::new(None, 1.to_string());
    let proc_handle = NodeHandle::new(Some(1), 1.to_string());
    let sink_handle = NodeHandle::new(Some(1), 3.to_string());

    dag.add_source(
        source_handle.clone(),
        Box::new(GeneratorSourceFactory::new(count, latch, false)),
    );
    dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));
    dag.add_sink(
        sink_handle.clone(),
        Box::new(ErrSinkFactory::new(200_000, false)),
    );

    dag.connect(
        Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),
        Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),
    )
    .unwrap();

    dag.connect(
        Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),
        Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),
    )
    .unwrap();

    DagExecutor::new(dag, ExecutorOptions::default())
        .unwrap()
        .start(Arc::new(AtomicBool::new(true)))
        .unwrap()
        .join()
        .unwrap();
}

#[test]
#[should_panic]
fn test_run_dag_sink_err_panic() {
    let count: u64 = 1_000_000;

    let mut dag = Dag::new();
    let latch = Arc::new(AtomicBool::new(true));

    let source_handle = NodeHandle::new(None, 1.to_string());
    let proc_handle = NodeHandle::new(Some(1), 1.to_string());
    let sink_handle = NodeHandle::new(Some(1), 3.to_string());

    dag.add_source(
        source_handle.clone(),
        Box::new(GeneratorSourceFactory::new(count, latch, false)),
    );
    dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));
    dag.add_sink(
        sink_handle.clone(),
        Box::new(ErrSinkFactory::new(200_000, true)),
    );

    dag.connect(
        Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),
        Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),
    )
    .unwrap();

    dag.connect(
        Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),
        Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),
    )
    .unwrap();

    DagExecutor::new(dag, ExecutorOptions::default())
        .unwrap()
        .start(Arc::new(AtomicBool::new(true)))
        .unwrap()
        .join()
        .unwrap();
}

1	use crate::channels::{ProcessorChannelForwarder, SourceChannelForwarder};
2	use crate::executor::{DagExecutor, ExecutorOptions};
3	use crate::executor_operation::ProcessorOperation;
4	use crate::node::{
5	OutputPortDef, OutputPortType, PortHandle, Processor, ProcessorFactory, Sink, SinkFactory,
6	Source, SourceFactory,
7	};
8	use crate::tests::dag_base_run::NoopProcessorFactory;
9	use crate::tests::sinks::{CountingSinkFactory, COUNTING_SINK_INPUT_PORT};
10	use crate::tests::sources::{GeneratorSourceFactory, GENERATOR_SOURCE_OUTPUT_PORT};
11	use crate::{Dag, Endpoint, DEFAULT_PORT_HANDLE};
12	use dozer_types::epoch::Epoch;
13	use dozer_types::errors::internal::BoxedError;
14	use dozer_types::ingestion_types::IngestionMessage;
15	use dozer_types::node::NodeHandle;
16	use dozer_types::types::{
17	Field, FieldDefinition, FieldType, Operation, Record, Schema, SourceDefinition,
18	};
19
20	use std::collections::HashMap;
21	use std::panic;
22
23	use std::sync::atomic::AtomicBool;
24	use std::sync::Arc;
25
26	use crate::tests::app::NoneContext;
27
28	// Test when error is generated by a processor
29
30	#[derive(Debug)]	×
31	struct ErrorProcessorFactory {
32	err_on: u64,
33	panic: bool,
34	}
35
36	impl ProcessorFactory<NoneContext> for ErrorProcessorFactory {
37	fn type_name(&self) -> String {	×
38	"Error".to_owned()	×
39	}	×
40
41	fn get_output_schema(	3✔
42	&self,	3✔
43	_output_port: &PortHandle,	3✔
44	input_schemas: &HashMap<PortHandle, (Schema, NoneContext)>,	3✔
45	) -> Result<(Schema, NoneContext), BoxedError> {	3✔
46	Ok(input_schemas.get(&DEFAULT_PORT_HANDLE).unwrap().clone())	3✔
47	}	3✔
48
49	fn get_input_ports(&self) -> Vec<PortHandle> {	12✔
50	vec![DEFAULT_PORT_HANDLE]	12✔
51	}	12✔
52
53	fn get_output_ports(&self) -> Vec<OutputPortDef> {	9✔
54	vec![OutputPortDef::new(	9✔
55	DEFAULT_PORT_HANDLE,	9✔
56	OutputPortType::Stateless,	9✔
57	)]	9✔
58	}	9✔
59
60	fn build(	3✔
61	&self,	3✔
62	_input_schemas: HashMap<PortHandle, Schema>,	3✔
63	_output_schemas: HashMap<PortHandle, Schema>,	3✔
64	) -> Result<Box<dyn Processor>, BoxedError> {	3✔
65	Ok(Box::new(ErrorProcessor {	3✔
66	err_on: self.err_on,	3✔
67	count: 0,	3✔
68	panic: self.panic,	3✔
69	}))	3✔
70	}	3✔
71
72	fn id(&self) -> String {	×
73	"Error".to_owned()	×
74	}	×
75	}
76
77	#[derive(Debug)]	×
78	struct ErrorProcessor {
79	err_on: u64,
80	count: u64,
81	panic: bool,
82	}
83
84	impl Processor for ErrorProcessor {
85	fn commit(&self, _epoch: &Epoch) -> Result<(), BoxedError> {	249✔
86	Ok(())	249✔
87	}	249✔
88
89	fn process(	2,400,000✔
90	&mut self,	2,400,000✔
91	_from_port: PortHandle,	2,400,000✔
92	op: ProcessorOperation,	2,400,000✔
93	fw: &mut dyn ProcessorChannelForwarder,	2,400,000✔
94	) -> Result<(), BoxedError> {	2,400,000✔
95	self.count += 1;	2,400,000✔
96	if self.count == self.err_on {	2,400,000✔
97	if self.panic {	3✔
98	panic!("Generated error");	1✔
99	} else {
100	return Err("Uknown".to_string().into());	2✔
101	}
102	}	2,399,997✔
103		2,399,997✔
104	fw.send(op, DEFAULT_PORT_HANDLE);	2,399,997✔
105	Ok(())	2,399,997✔
106	}	2,399,999✔
107	}
108
109	#[test]	1✔
110	#[should_panic]
111	fn test_run_dag_proc_err_panic() {	1✔
112	let count: u64 = 1_000_000;	1✔
113		1✔
114	let mut dag = Dag::new();	1✔
115	let latch = Arc::new(AtomicBool::new(true));	1✔
116		1✔
117	let source_handle = NodeHandle::new(None, 1.to_string());	1✔
118	let proc_handle = NodeHandle::new(Some(1), 1.to_string());	1✔
119	let sink_handle = NodeHandle::new(Some(1), 2.to_string());	1✔
120		1✔
121	dag.add_source(	1✔
122	source_handle.clone(),	1✔
123	Box::new(GeneratorSourceFactory::new(count, latch.clone(), false)),	1✔
124	);	1✔
125	dag.add_processor(	1✔
126	proc_handle.clone(),	1✔
127	Box::new(ErrorProcessorFactory {	1✔
128	err_on: 800_000,	1✔
129	panic: true,	1✔
130	}),	1✔
131	);	1✔
132	dag.add_sink(	1✔
133	sink_handle.clone(),	1✔
134	Box::new(CountingSinkFactory::new(count, latch)),	1✔
135	);	1✔
136		1✔
137	dag.connect(	1✔
138	Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),	1✔
139	Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),	1✔
140	)	1✔
141	.unwrap();	1✔
142		1✔
143	dag.connect(	1✔
144	Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),	1✔
145	Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),	1✔
146	)	1✔
147	.unwrap();	1✔
148		1✔
149	DagExecutor::new(dag, ExecutorOptions::default())	1✔
150	.unwrap()	1✔
151	.start(Arc::new(AtomicBool::new(true)))	1✔
152	.unwrap()	1✔
153	.join()	1✔
154	.unwrap();	1✔
155	}	1✔
156
157	#[test]	1✔
158	#[should_panic]
159	fn test_run_dag_proc_err_2() {	1✔
160	let count: u64 = 1_000_000;	1✔
161		1✔
162	let mut dag = Dag::new();	1✔
163	let latch = Arc::new(AtomicBool::new(true));	1✔
164		1✔
165	let source_handle = NodeHandle::new(None, 1.to_string());	1✔
166	let proc_handle = NodeHandle::new(Some(1), 1.to_string());	1✔
167	let proc_err_handle = NodeHandle::new(Some(1), 2.to_string());	1✔
168	let sink_handle = NodeHandle::new(Some(1), 3.to_string());	1✔
169		1✔
170	dag.add_source(	1✔
171	source_handle.clone(),	1✔
172	Box::new(GeneratorSourceFactory::new(count, latch.clone(), false)),	1✔
173	);	1✔
174	dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));	1✔
175		1✔
176	dag.add_processor(	1✔
177	proc_err_handle.clone(),	1✔
178	Box::new(ErrorProcessorFactory {	1✔
179	err_on: 800_000,	1✔
180	panic: false,	1✔
181	}),	1✔
182	);	1✔
183		1✔
184	dag.add_sink(	1✔
185	sink_handle.clone(),	1✔
186	Box::new(CountingSinkFactory::new(count, latch)),	1✔
187	);	1✔
188		1✔
189	dag.connect(	1✔
190	Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),	1✔
191	Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),	1✔
192	)	1✔
193	.unwrap();	1✔
194		1✔
195	dag.connect(	1✔
196	Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),	1✔
197	Endpoint::new(proc_err_handle.clone(), DEFAULT_PORT_HANDLE),	1✔
198	)	1✔
199	.unwrap();	1✔
200		1✔
201	dag.connect(	1✔
202	Endpoint::new(proc_err_handle, DEFAULT_PORT_HANDLE),	1✔
203	Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),	1✔
204	)	1✔
205	.unwrap();	1✔
206		1✔
207	DagExecutor::new(dag, ExecutorOptions::default())	1✔
208	.unwrap()	1✔
209	.start(Arc::new(AtomicBool::new(true)))	1✔
210	.unwrap()	1✔
211	.join()	1✔
212	.unwrap();	1✔
213	}	1✔
214
215	#[test]	1✔
216	#[should_panic]
217	fn test_run_dag_proc_err_3() {	1✔
218	let count: u64 = 1_000_000;	1✔
219		1✔
220	let mut dag = Dag::new();	1✔
221	let latch = Arc::new(AtomicBool::new(true));	1✔
222		1✔
223	let source_handle = NodeHandle::new(None, 1.to_string());	1✔
224	let proc_handle = NodeHandle::new(Some(1), 1.to_string());	1✔
225	let proc_err_handle = NodeHandle::new(Some(1), 2.to_string());	1✔
226	let sink_handle = NodeHandle::new(Some(1), 3.to_string());	1✔
227		1✔
228	dag.add_source(	1✔
229	source_handle.clone(),	1✔
230	Box::new(GeneratorSourceFactory::new(count, latch.clone(), false)),	1✔
231	);	1✔
232		1✔
233	dag.add_processor(	1✔
234	proc_err_handle.clone(),	1✔
235	Box::new(ErrorProcessorFactory {	1✔
236	err_on: 800_000,	1✔
237	panic: false,	1✔
238	}),	1✔
239	);	1✔
240		1✔
241	dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));	1✔
242		1✔
243	dag.add_sink(	1✔
244	sink_handle.clone(),	1✔
245	Box::new(CountingSinkFactory::new(count, latch)),	1✔
246	);	1✔
247		1✔
248	dag.connect(	1✔
249	Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),	1✔
250	Endpoint::new(proc_err_handle.clone(), DEFAULT_PORT_HANDLE),	1✔
251	)	1✔
252	.unwrap();	1✔
253		1✔
254	dag.connect(	1✔
255	Endpoint::new(proc_err_handle, DEFAULT_PORT_HANDLE),	1✔
256	Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),	1✔
257	)	1✔
258	.unwrap();	1✔
259		1✔
260	dag.connect(	1✔
261	Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),	1✔
262	Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),	1✔
263	)	1✔
264	.unwrap();	1✔
265		1✔
266	DagExecutor::new(dag, ExecutorOptions::default())	1✔
267	.unwrap()	1✔
268	.start(Arc::new(AtomicBool::new(true)))	1✔
269	.unwrap()	1✔
270	.join()	1✔
271	.unwrap();	1✔
272	}	1✔
273
274	// Test when error is generated by a source
275
276	#[derive(Debug)]	×
277	pub(crate) struct ErrGeneratorSourceFactory {
278	count: u64,
279	err_at: u64,
280	}
281
282	impl ErrGeneratorSourceFactory {
283	pub fn new(count: u64, err_at: u64) -> Self {	1✔
284	Self { count, err_at }	1✔
285	}	1✔
286	}
287
288	impl SourceFactory<NoneContext> for ErrGeneratorSourceFactory {
289	fn get_output_schema(&self, _port: &PortHandle) -> Result<(Schema, NoneContext), BoxedError> {	1✔
290	Ok((	1✔
291	Schema::default()	1✔
292	.field(	1✔
293	FieldDefinition::new(	1✔
294	"id".to_string(),	1✔
295	FieldType::String,	1✔
296	false,	1✔
297	SourceDefinition::Dynamic,	1✔
298	),	1✔
299	true,	1✔
300	)	1✔
301	.field(	1✔
302	FieldDefinition::new(	1✔
303	"value".to_string(),	1✔
304	FieldType::String,	1✔
305	false,	1✔
306	SourceDefinition::Dynamic,	1✔
307	),	1✔
308	false,	1✔
309	)	1✔
310	.clone(),	1✔
311	NoneContext {},	1✔
312	))	1✔
313	}	1✔
314
315	fn get_output_port_name(&self, _port: &PortHandle) -> String {	×
316	"error".to_string()	×
317	}	×
318		×
319	fn get_output_ports(&self) -> Vec<OutputPortDef> {	3✔
320	vec![OutputPortDef::new(	3✔
321	GENERATOR_SOURCE_OUTPUT_PORT,	3✔
322	OutputPortType::Stateless,	3✔
323	)]	3✔
324	}	3✔
325		×
326	fn build(	1✔
327	&self,	1✔
328	_output_schemas: HashMap<PortHandle, Schema>,	1✔
329	) -> Result<Box<dyn Source>, BoxedError> {	1✔
330	Ok(Box::new(ErrGeneratorSource {	1✔
331	count: self.count,	1✔
332	err_at: self.err_at,	1✔
333	}))	1✔
334	}	1✔
335	}
336
337	#[derive(Debug)]	×
338	pub(crate) struct ErrGeneratorSource {
339	count: u64,
340	err_at: u64,	×
341	}	×
342		×
343	impl Source for ErrGeneratorSource {
344	fn can_start_from(&self, _last_checkpoint: (u64, u64)) -> Result<bool, BoxedError> {	×
345	Ok(false)	×
346	}	×
347		×
348	fn start(	1✔
349	&self,	1✔
350	fw: &mut dyn SourceChannelForwarder,	1✔
351	_checkpoint: Option<(u64, u64)>,	1✔
352	) -> Result<(), BoxedError> {	1✔
353	for n in 1..(self.count + 1) {	200,000✔
354	if n == self.err_at {	200,000✔
355	return Err("Generated Error".to_string().into());	1✔
356	}	199,999✔
357		199,999✔
358	fw.send(	199,999✔
359	IngestionMessage::new_op(	199,999✔
360	n,	199,999✔
361	0,	199,999✔
362	0,	199,999✔
363	Operation::Insert {	199,999✔
364	new: Record::new(vec![	199,999✔
365	Field::String(format!("key_{n}")),	199,999✔
366	Field::String(format!("value_{n}")),	199,999✔
367	]),	199,999✔
368	},	199,999✔
369	),	199,999✔
370	GENERATOR_SOURCE_OUTPUT_PORT,	199,999✔
371	)?;	199,999✔
372	}
373	Ok(())	×
374	}	1✔
375	}	×
376		×
377	#[test]	1✔
378	fn test_run_dag_src_err() {	1✔
379	let count: u64 = 1_000_000;	1✔
380		1✔
381	let mut dag = Dag::new();	1✔
382	let latch = Arc::new(AtomicBool::new(true));	1✔
383		1✔
384	let source_handle = NodeHandle::new(None, 1.to_string());	1✔
385	let proc_handle = NodeHandle::new(Some(1), 1.to_string());	1✔
386	let sink_handle = NodeHandle::new(Some(1), 3.to_string());	1✔
387		1✔
388	dag.add_source(	1✔
389	source_handle.clone(),	1✔
390	Box::new(ErrGeneratorSourceFactory::new(count, 200_000)),	1✔
391	);	1✔
392	dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));	1✔
393	dag.add_sink(	1✔
394	sink_handle.clone(),	1✔
395	Box::new(CountingSinkFactory::new(count, latch)),	1✔
396	);	1✔
397		1✔
398	dag.connect(	1✔
399	Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),	1✔
400	Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),	1✔
401	)	1✔
402	.unwrap();	1✔
403		1✔
404	dag.connect(	1✔
405	Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),	1✔
406	Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),	1✔
407	)	1✔
408	.unwrap();	1✔
409		1✔
410	let _join_handle = DagExecutor::new(dag, ExecutorOptions::default())	1✔
411	.unwrap()	1✔
412	.start(Arc::new(AtomicBool::new(true)))	1✔
413	.unwrap();	1✔
414	// join_handle.join().unwrap();	1✔
415	}	1✔
416
417	#[derive(Debug)]	×
418	pub(crate) struct ErrSinkFactory {
419	err_at: u64,
420	panic: bool,	×
421	}	×
422		×
423	impl ErrSinkFactory {
424	pub fn new(err_at: u64, panic: bool) -> Self {	2✔
425	Self { err_at, panic }	2✔
426	}	2✔
427	}	×
428		×
429	impl SinkFactory<NoneContext> for ErrSinkFactory {
430	fn get_input_ports(&self) -> Vec<PortHandle> {	8✔
431	vec![COUNTING_SINK_INPUT_PORT]	8✔
432	}	8✔
433		×
434	fn prepare(	2✔
435	&self,	2✔
436	_input_schemas: HashMap<PortHandle, (Schema, NoneContext)>,	2✔
437	) -> Result<(), BoxedError> {	2✔
438	Ok(())	2✔
439	}	2✔
440		×
441	fn build(	2✔
442	&self,	2✔
443	_input_schemas: HashMap<PortHandle, Schema>,	2✔
444	) -> Result<Box<dyn Sink>, BoxedError> {	2✔
445	Ok(Box::new(ErrSink {	2✔
446	err_at: self.err_at,	2✔
447	current: 0,	2✔
448	panic: self.panic,	2✔
449	}))	2✔
450	}	2✔
451	}
452
453	#[derive(Debug)]	×
454	pub(crate) struct ErrSink {
455	err_at: u64,
456	current: u64,	×
457	panic: bool,	×
458	}	×
459	impl Sink for ErrSink {
460	fn commit(&mut self, _epoch_details: &Epoch) -> Result<(), BoxedError> {	42✔
461	Ok(())	42✔
462	}	42✔
463		×
464	fn process(	400,000✔
465	&mut self,	400,000✔
466	_from_port: PortHandle,	400,000✔
467	_op: ProcessorOperation,	400,000✔
468	) -> Result<(), BoxedError> {	400,000✔
469	self.current += 1;	400,000✔
470	if self.current == self.err_at {	400,000✔
471	if self.panic {	2✔
472	panic!("Generated error");	1✔
473	} else {	×
474	return Err("Generated error".to_string().into());	1✔
475	}
476	}	399,998✔
477	Ok(())	399,998✔
478	}	399,999✔
479
480	fn on_source_snapshotting_done(&mut self, _connection_name: String) -> Result<(), BoxedError> {	×
481	Ok(())	×
482	}	×
483	}	×
484		×
485	#[test]	1✔
486	#[should_panic]	×
487	fn test_run_dag_sink_err() {	1✔
488	let count: u64 = 1_000_000;	1✔
489		1✔
490	let mut dag = Dag::new();	1✔
491	let latch = Arc::new(AtomicBool::new(true));	1✔
492		1✔
493	let source_handle = NodeHandle::new(None, 1.to_string());	1✔
494	let proc_handle = NodeHandle::new(Some(1), 1.to_string());	1✔
495	let sink_handle = NodeHandle::new(Some(1), 3.to_string());	1✔
496		1✔
497	dag.add_source(	1✔
498	source_handle.clone(),	1✔
499	Box::new(GeneratorSourceFactory::new(count, latch, false)),	1✔
500	);	1✔
501	dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));	1✔
502	dag.add_sink(	1✔
503	sink_handle.clone(),	1✔
504	Box::new(ErrSinkFactory::new(200_000, false)),	1✔
505	);	1✔
506		1✔
507	dag.connect(	1✔
508	Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),	1✔
509	Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),	1✔
510	)	1✔
511	.unwrap();	1✔
512		1✔
513	dag.connect(	1✔
514	Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),	1✔
515	Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),	1✔
516	)	1✔
517	.unwrap();	1✔
518		1✔
519	DagExecutor::new(dag, ExecutorOptions::default())	1✔
520	.unwrap()	1✔
521	.start(Arc::new(AtomicBool::new(true)))	1✔
522	.unwrap()	1✔
523	.join()	1✔
524	.unwrap();	1✔
525	}	1✔
526		×
527	#[test]	1✔
528	#[should_panic]	×
529	fn test_run_dag_sink_err_panic() {	1✔
530	let count: u64 = 1_000_000;	1✔
531		1✔
532	let mut dag = Dag::new();	1✔
533	let latch = Arc::new(AtomicBool::new(true));	1✔
534		1✔
535	let source_handle = NodeHandle::new(None, 1.to_string());	1✔
536	let proc_handle = NodeHandle::new(Some(1), 1.to_string());	1✔
537	let sink_handle = NodeHandle::new(Some(1), 3.to_string());	1✔
538		1✔
539	dag.add_source(	1✔
540	source_handle.clone(),	1✔
541	Box::new(GeneratorSourceFactory::new(count, latch, false)),	1✔
542	);	1✔
543	dag.add_processor(proc_handle.clone(), Box::new(NoopProcessorFactory {}));	1✔
544	dag.add_sink(	1✔
545	sink_handle.clone(),	1✔
546	Box::new(ErrSinkFactory::new(200_000, true)),	1✔
547	);	1✔
548		1✔
549	dag.connect(	1✔
550	Endpoint::new(source_handle, GENERATOR_SOURCE_OUTPUT_PORT),	1✔
551	Endpoint::new(proc_handle.clone(), DEFAULT_PORT_HANDLE),	1✔
552	)	1✔
553	.unwrap();	1✔
554		1✔
555	dag.connect(	1✔
556	Endpoint::new(proc_handle, DEFAULT_PORT_HANDLE),	1✔
557	Endpoint::new(sink_handle, COUNTING_SINK_INPUT_PORT),	1✔
558	)	1✔
559	.unwrap();	1✔
560		1✔
561	DagExecutor::new(dag, ExecutorOptions::default())	1✔
562	.unwrap()	1✔
563	.start(Arc::new(AtomicBool::new(true)))	1✔
564	.unwrap()	1✔
565	.join()	1✔
566	.unwrap();	1✔
567	}	1✔

getdozer / dozer / 5725710489

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous