ref: Upgrade arroyo, use coarsetime (#5934)

* ref: Upgrade arroyo, use coarsetime

Changes necessary for getsentry/arroyo#366, and
additional timers for get_str_config, since that one acquires multiple
global locks and might impact concurrency.

* bump arroyo

* update arroyo again

* remove another close

* fix tests again

rust_snuba/bin/python_processor_infinite.rs

@@ -51,7 +51,6 @@ fn main() {
 
     println!("join");
 
-    step.close();
     step.join(None).unwrap();
 
     println!("{}", output.load(Ordering::Relaxed))

rust_snuba/src/factory.rs

@@ -67,7 +67,7 @@ impl ProcessingStrategyFactory<KafkaPayload> for ConsumerStrategyFactory {
 
     fn create(&self) -> Box<dyn ProcessingStrategy<KafkaPayload>> {
         // Commit offsets
-        let next_step = CommitOffsets::new(chrono::Duration::seconds(1));
+        let next_step = CommitOffsets::new(Duration::from_secs(1));
 
         // Produce commit log if there is one
         let next_step: Box<dyn ProcessingStrategy<BytesInsertBatch<()>>> =
@@ -119,8 +119,9 @@ impl ProcessingStrategyFactory<KafkaPayload> for ConsumerStrategyFactory {
         );
 
         let accumulator = Arc::new(
-            |batch: BytesInsertBatch<HttpBatch>, small_batch: BytesInsertBatch<RowData>| {
-                batch.merge(small_batch)
+            |batch: BytesInsertBatch<HttpBatch>,
+             small_batch: Message<BytesInsertBatch<RowData>>| {
+                Ok(batch.merge(small_batch.into_payload()))
             },
         );
 

rust_snuba/src/runtime_config.rs

@@ -4,13 +4,12 @@ use pyo3::prelude::{PyModule, Python};
 use std::collections::BTreeMap;
 use std::time::Duration;
 
+use rust_arroyo::timer;
 use rust_arroyo::utils::timing::Deadline;
 
-#[allow(dead_code)]
 static CONFIG: RwLock<BTreeMap<String, (Option<String>, Deadline)>> = RwLock::new(BTreeMap::new());
 
 /// Runtime config is cached for 10 seconds
-#[allow(dead_code)]
 pub fn get_str_config(key: &str) -> Result<Option<String>, Error> {
     let deadline = Deadline::new(Duration::from_secs(10));
 
@@ -21,7 +20,7 @@ pub fn get_str_config(key: &str) -> Result<Option<String>, Error> {
         }
     }
 
-    Python::with_gil(|py| {
+    let rv = Python::with_gil(|py| {
         let snuba_state = PyModule::import(py, "snuba.state")?;
         let config = snuba_state
             .getattr("get_str_config")?
@@ -32,7 +31,11 @@ pub fn get_str_config(key: &str) -> Result<Option<String>, Error> {
             .write()
             .insert(key.to_string(), (config.clone(), deadline));
         Ok(CONFIG.read().get(key).unwrap().0.clone())
-    })
+    });
+
+    timer!("runtime_config.get_str_config", deadline.elapsed());
+
+    rv
 }
 
 #[cfg(test)]

rust_snuba/src/strategies/accountant.rs

@@ -83,10 +83,6 @@ where
         self.next_step.submit(message)
     }
 
-    fn close(&mut self) {
-        self.next_step.close()
-    }
-
     fn terminate(&mut self) {
         self.next_step.terminate()
     }

rust_snuba/src/strategies/clickhouse/mod.rs

@@ -99,10 +99,6 @@ impl ProcessingStrategy<BytesInsertBatch<HttpBatch>> for ClickhouseWriterStep {
         self.inner.submit(message)
     }
 
-    fn close(&mut self) {
-        self.inner.close();
-    }
-
     fn terminate(&mut self) {
         self.inner.terminate();
     }

rust_snuba/src/strategies/commit_log.rs

@@ -185,10 +185,6 @@ impl ProcessingStrategy<BytesInsertBatch<()>> for ProduceCommitLog {
         self.inner.submit(message)
     }
 
-    fn close(&mut self) {
-        self.inner.close();
-    }
-
     fn terminate(&mut self) {
         self.inner.terminate();
     }
@@ -353,7 +349,6 @@ mod tests {
             strategy.poll().unwrap();
         }
 
-        strategy.close();
         strategy.join(None).unwrap();
 
         let produced = produced_payloads.lock().unwrap();

rust_snuba/src/strategies/join_timeout.rs

@@ -35,10 +35,6 @@ where
         self.next_step.terminate()
     }
 
-    fn close(&mut self) {
-        self.next_step.close();
-    }
-
     fn join(&mut self, _timeout: Option<Duration>) -> Result<Option<CommitRequest>, StrategyError> {
         self.next_step.join(self.new_timeout)
     }

rust_snuba/src/strategies/noop.rs

@@ -16,8 +16,6 @@ impl<T> ProcessingStrategy<T> for Noop {
         Ok(())
     }
 
-    fn close(&mut self) {}
-
     fn terminate(&mut self) {}
 
     fn join(&mut self, _timeout: Option<Duration>) -> Result<Option<CommitRequest>, StrategyError> {

rust_snuba/src/strategies/processor.rs

@@ -384,7 +384,6 @@ mod tests {
         let message = Message::new_broker_message(payload, partition, 0, Utc::now());
 
         strategy.submit(message).unwrap(); // Does not error
-        strategy.close();
         let _ = strategy.join(None);
     }
 }

rust_snuba/src/strategies/python.rs

@@ -215,8 +215,6 @@ impl ProcessingStrategy<KafkaPayload> for PythonTransformStep {
         Ok(())
     }
 
-    fn close(&mut self) {}
-
     fn terminate(&mut self) {
         self.next_step.terminate()
     }
@@ -260,8 +258,8 @@ impl ProcessingStrategy<KafkaPayload> for PythonTransformStep {
             }
         }
 
-        self.next_step.close();
         let next_commit = self.next_step.join(timeout)?;
+
         Ok(merge_commit_request(
             self.commit_request_carried_over.take(),
             next_commit,

rust_snuba/src/strategies/replacements.rs

@@ -105,11 +105,6 @@ impl ProcessingStrategy<InsertOrReplacement<BytesInsertBatch<RowData>>> for Prod
         }
     }
 
-    fn close(&mut self) {
-        self.inner.close();
-        self.next_step.close();
-    }
-
     fn terminate(&mut self) {
         self.inner.terminate();
         self.next_step.terminate();
@@ -178,7 +173,6 @@ mod tests {
             .unwrap();
 
         strategy.poll().unwrap();
-        strategy.close();
         strategy.join(None).unwrap();
 
         assert_eq!(produced_payloads.lock().unwrap().len(), 1);

rust_snuba/src/testutils.rs

@@ -50,7 +50,6 @@ impl<R: Send + Sync> ProcessingStrategy<BytesInsertBatch<R>> for TestStrategy<R>
         self.payloads.push(message.into_payload());
         Ok(())
     }
-    fn close(&mut self) {}
     fn terminate(&mut self) {}
     fn join(&mut self, _timeout: Option<Duration>) -> Result<Option<CommitRequest>, StrategyError> {
         Ok(None)