Fix wal config
@@ -140,8 +140,7 @@ pub struct Config {
pub max_bytes: u64,
/// Flush to disk on every write
- #[serde(default = "Default::default")]
- flush_on_evnt: bool,
+ flush_on_evnt: Option<bool>,
}
impl ConfigImpl for Config {}
@@ -265,7 +264,7 @@ impl WAL {
// Sieralize and write the event
let event_buf = event.json_vec()?;
self.events_tree.insert(write, event_buf.as_slice())?;
- if self.config.flush_on_evnt {
+ if self.config.flush_on_evnt.unwrap_or_default() {
self.events_tree.flush()?;
}
self.cnt += 1;
@@ -420,6 +419,7 @@ mod test {
dir: None,
max_elements: 10,
max_bytes: 1024 * 1024,
+ flush_on_evnt: None,
};
let mut o = WalFactory::new().from_node(&NodeConfig::from_config("test", c)?)?;
let mut v = Value::null();
@@ -470,4 +470,64 @@ mod test {
Ok(())
}
+
+ #[test]
+ // tests that the wal works fine
+ // after a restart of the tremor server
+ fn restart_wal_regression() -> Result<()> {
+ let temp_dir = TempDirBuilder::new()
+ .prefix("tremor-pipeline-wal")
+ .tempdir()?;
+ let read_count = 100;
+ let c = Config {
+ read_count,
+ dir: Some(temp_dir.path().to_string_lossy().into_owned()),
+ max_elements: 10,
+ max_bytes: 1024 * 1024,
+ flush_on_evnt: None,
+ };
+
+ let mut v = Value::null();
+ let e = Event::default();
+
+ {
+ // create the operator - first time
+ let mut o1 =
+ WalFactory::new().from_node(&NodeConfig::from_config("wal-test-1", c.clone())?)?;
+
+ // Restore the CB
+ let mut i = Event::cb_restore(0);
+ o1.on_contraflow(0, &mut i);
+
+ // send a first event - not acked. so it lingers around in our WAL
+ let r = o1.on_event(0, "in", &mut v, e.clone())?;
+ assert_eq!(r.events.len(), 1);
+ assert_eq!(r.insights.len(), 0);
+ }
+
+ {
+ // create the operator - second time
+ // simulating a tremor restart
+ let mut o2 = WalFactory::new().from_node(&NodeConfig::from_config("wal-test-2", c)?)?;
+
+ // Restore the CB
+ let mut i = Event::cb_restore(1);
+ o2.on_contraflow(0, &mut i);
+
+ // send a first event - not acked. so it lingers around in our
+ let r = o2.on_event(0, "in", &mut v, e.clone())?;
+ assert_eq!(r.events.len(), 2);
+ let id1 = &r.events[0].1.id;
+ let id2 = &r.events[1].1.id;
+ assert_eq!(id1.get(0).unwrap(), 0);
+ // ensure we actually had a gap bigger than read count, which triggers the error condition
+ assert!(id2.get(0).unwrap() - id1.get(0).unwrap() > read_count as u64);
+ assert_eq!(r.insights.len(), 0);
+
+ let r = o2.on_event(0, "in", &mut v, e.clone())?;
+ assert_eq!(r.events.len(), 1);
+ }
+
+ Ok(())
+ }
}
