wip(taskbroker): Add consumer step to resend delayed task to kafka #297
Conversation
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #297 +/- ##
==========================================
- Coverage 84.30% 82.22% -2.08%
==========================================
Files 20 21 +1
Lines 3771 3866 +95
==========================================
Hits 3179 3179
- Misses 592 687 +95 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
john-z-yang
force-pushed
the
john/reroute-delay-task
branch
from
c4367d5 to
1aa01ea
Compare
john-z-yang
force-pushed
the
john/reroute-delay-task
branch
from
1aa01ea to
9c42502
Compare
john-z-yang
changed the title
| store: Arc<InflightActivationStore>, | ||
| batch: Option<Vec<InflightActivation>>, | ||
| } | ||
|
|
||
| impl InflightActivationWriter { | ||
| pub fn new(store: Arc<InflightActivationStore>, config: ActivationWriterConfig) -> Self { | ||
| impl ActivationStoreWriter { |
There was a problem hiding this comment.
Maybe we should deprecate 'inflightactivation' 🤔 we'll need a term to disambiguate the protobuf form and the sqlite mapped struct though.
| let (ready_activations, delayed_activations): (Vec<_>, Vec<_>) = take(&mut self.batch) | ||
| .unwrap() | ||
| .into_iter() | ||
| .partition(|_| true); |
There was a problem hiding this comment.
We'll need a field in the protobuf to make this work for real. Options I've thought of are countdown and runafter 🤷
| let deliveries = payloads | ||
| .into_iter() | ||
| .map(|payload| { | ||
| let producer = self.producer.clone(); | ||
| let config = self.config.clone(); | ||
| async move { | ||
| producer | ||
| .send( | ||
| FutureRecord::<(), Vec<u8>>::to(&config.kafka_topic) | ||
| .payload(&payload), | ||
| Timeout::After(Duration::from_millis(config.kafka_send_timeout_ms)), | ||
| ) | ||
| .await | ||
| } | ||
| }) | ||
| .collect::<FuturesUnordered<_>>(); |
There was a problem hiding this comment.
We have this pattern in a few places now, should we try making a function for it?
There was a problem hiding this comment.
We should, they're all slightly different tho. So we need to be clever with our abstraction.
TODO: