change return type of runtask

getsentry · Oct 17, 2024 · 45eef29 · 45eef29
1 parent c475f0c
commit 45eef29
Show file tree

Hide file tree

Showing 3 changed files with 9 additions and 8 deletions.
diff --git a/rust-arroyo/examples/transform_and_produce.rs b/rust-arroyo/examples/transform_and_produce.rs
@@ -14,12 +14,14 @@ use rust_arroyo::processing::strategies::produce::Produce;
 use rust_arroyo::processing::strategies::run_task::RunTask;
 use rust_arroyo::processing::strategies::run_task_in_threads::ConcurrencyConfig;
 use rust_arroyo::processing::strategies::{
-    InvalidMessage, ProcessingStrategy, ProcessingStrategyFactory,
+    ProcessingStrategy, ProcessingStrategyFactory, SubmitError,
 };
 use rust_arroyo::processing::StreamProcessor;
 use rust_arroyo::types::{Message, Topic, TopicOrPartition};
 
-fn reverse_string(message: Message<KafkaPayload>) -> Result<Message<KafkaPayload>, InvalidMessage> {
+fn reverse_string(
+    message: Message<KafkaPayload>,
+) -> Result<Message<KafkaPayload>, SubmitError<KafkaPayload>> {
     let value = message.payload();
     let payload = value.payload().unwrap();
     let str_payload = std::str::from_utf8(payload).unwrap();

diff --git a/rust-arroyo/src/processing/strategies/commit_offsets.rs b/rust-arroyo/src/processing/strategies/commit_offsets.rs
@@ -95,7 +95,6 @@ mod tests {
     #[test]
     fn test_commit_offsets() {
         tracing_subscriber::fmt().with_test_writer().init();
-        let updater = coarsetime::Updater::new(10).start().unwrap();
 
         let partition1 = Partition::new(Topic::new("noop-commit"), 0);
         let partition2 = Partition::new(Topic::new("noop-commit"), 1);

diff --git a/rust-arroyo/src/processing/strategies/run_task.rs b/rust-arroyo/src/processing/strategies/run_task.rs
@@ -1,6 +1,6 @@
 use crate::processing::strategies::{
-    merge_commit_request, CommitRequest, InvalidMessage, MessageRejected, ProcessingStrategy,
-    StrategyError, SubmitError,
+    merge_commit_request, CommitRequest, MessageRejected, ProcessingStrategy, StrategyError,
+    SubmitError,
 };
 use crate::types::Message;
 use std::time::Duration;
@@ -26,7 +26,7 @@ impl<TTransformed, F, N> RunTask<TTransformed, F, N> {
 impl<TPayload, TTransformed, F, N> ProcessingStrategy<TPayload> for RunTask<TTransformed, F, N>
 where
     TTransformed: Send + Sync,
-    F: FnMut(Message<TPayload>) -> Result<Message<TTransformed>, InvalidMessage>
+    F: FnMut(Message<TPayload>) -> Result<Message<TTransformed>, SubmitError<TPayload>>
         + Send
         + Sync
         + 'static,
@@ -63,7 +63,7 @@ where
             return Err(SubmitError::MessageRejected(MessageRejected { message }));
         }
 
-        let next_message = (self.function)(message).map_err(SubmitError::InvalidMessage)?;
+        let next_message = (self.function)(message)?;
 
         match self.next_step.submit(next_message) {
             Err(SubmitError::MessageRejected(MessageRejected {
@@ -103,7 +103,7 @@ mod tests {
 
     #[test]
     fn test_run_task() {
-        fn identity(value: Message<String>) -> Result<Message<String>, InvalidMessage> {
+        fn identity(value: Message<String>) -> Result<Message<String>, SubmitError<String>> {
             Ok(value)
         }