Conversation
bmckerry
changed the title
bmckerry
force-pushed
the
74/broadcast-custom-step
branch
from
e354fa3 to
29a67bd
Compare
| @@ -1,50 +0,0 @@ | |||
| from sentry_streams.examples.broadcast_fn import BroadcastFunctions | |||
There was a problem hiding this comment.
deleting this as it isn't a practical example and implicit broadcast doesn't work now
| @@ -1,166 +1,167 @@ | |||
| version = 1 | |||
There was a problem hiding this comment.
not sure what happened here, I thought this diff was already on main
|
|
||
| impl ProcessingStrategy<RoutedValue> for Broadcaster { | ||
| fn poll(&mut self) -> Result<Option<CommitRequest>, StrategyError> { | ||
| self.flush_pending()?; |
There was a problem hiding this comment.
This is a situation I'm not sure about:
- a watermark message goes thru the broadcast step with 2 downstream branches, the first copy receives a
MessageRejectedbackpressure error and goes into the pending buffer - before the watermark message is retried,
poll()is called which flushes the pending buffer, sending the previously failed message copy downstream- commit step receives watermark from branch 1
- the failed message is retried and successfully passes through the broadcast step, sending 2 more copies of the message downstream
- commit step receives watermark from branches 1 and 2
In the above situation we end up with 3 watermarks reaching the commit step when the commit step only expects 2 (since there's 2 downstream routes), meaning if we're just counting watermarks we'll commit after we receive the 2nd watermark copy from branch 1 without receiving the watermark copy from branch 2.
Questions:
- can arroyo call
poll()between when a message receivesMessageRejectedand when that message is retried? - does this mean that in the custom commit step, we'll need to ensure we receive a watermark with a specific
committablepayload routed for each possible branch, and not just count the number of watermarks with a givencommittablepayload?
There was a problem hiding this comment.
we'll need to ensure we receive a watermark with a specific committable payload routed for each possible branch, and not just count the number of watermarks with a given committable payload?
talked with @evanh and this is basically the answer to this problem
There was a problem hiding this comment.
I am not sure I understand the problem.
Why would we ever send a second copy of the watermark ?
If we try to send a watermark, route 1 works and route 2 receives a MessageRejected:
- The next poll should only re-send route 2 watermark.
- route 1 watermak should not be sent again as you do for any other message.
Are we not using the message identifier to avoid sending another copy of the same watermark ?
| impl Clone for PyStreamingMessage { | ||
| fn clone(&self) -> Self { | ||
| match self { | ||
| PyStreamingMessage::PyAnyMessage { ref content } => { | ||
| let py_any_clone = traced_with_gil!(|py| content.clone_ref(py)); | ||
| PyStreamingMessage::PyAnyMessage { | ||
| content: py_any_clone, | ||
| } | ||
| } | ||
| PyStreamingMessage::RawMessage { ref content } => { | ||
| let raw_clone = traced_with_gil!(|py| content.clone_ref(py)); | ||
| PyStreamingMessage::RawMessage { content: raw_clone } | ||
| } | ||
| } |
There was a problem hiding this comment.
This is not creating a clone of the message.
It is only cloning the reference to the Python payload, which means that, if the payload is mutable, two branches may update the same message.
The only way to clone the payload is to relying on Python code to clone it (maybe using deepcopy).
In these scenarios unit tests can help you to validate correctness of your code. The identity of the python object of your clone would not change.
| let unfolded_messages = generate_broadcast_messages(&self.downstream_branches, message); | ||
| for msg in unfolded_messages { | ||
| self.handle_submit(msg)?; | ||
| } |
There was a problem hiding this comment.
I don't think this works as expected.
As this piece of code does not flush the pending messages, if there are pending messages, you would submit the newer message before the pending messages (that are older) thus sending messages out of order, which is wrong.
Please ensure this case is covered by your unit tests.
|
|
||
| impl ProcessingStrategy<RoutedValue> for Broadcaster { | ||
| fn poll(&mut self) -> Result<Option<CommitRequest>, StrategyError> { | ||
| self.flush_pending()?; |
There was a problem hiding this comment.
I am not sure I understand the problem.
Why would we ever send a second copy of the watermark ?
If we try to send a watermark, route 1 works and route 2 receives a MessageRejected:
- The next poll should only re-send route 2 watermark.
- route 1 watermak should not be sent again as you do for any other message.
Are we not using the message identifier to avoid sending another copy of the same watermark ?
| let _ = step.submit(message.clone()); | ||
| assert_eq!(step.pending_messages.len(), 1); | ||
| let actual_messages = submitted_messages_clone.lock().unwrap(); | ||
| assert_messages_match( | ||
| py, | ||
| vec![ | ||
| "test_message".into_py_any(py).unwrap(), | ||
| "test_message".into_py_any(py).unwrap(), | ||
| ], | ||
| actual_messages.deref(), |
There was a problem hiding this comment.
This is not enough to test the functionality. It does not check messages are sent on all routes.
3 participants
This PR:
Broadcaststep which makes a copy of a message for each downstream route, where each copy is routed to one of those branchesClonetrait forRoutedValue(and conversely for each variant ofRoutedValuePayload)uv.lockdiff for some reasonMy plan going forward is to re-use the logic from
generate_broadcast_messages()when making a custom Router step that routes regular messages downstream (to a single downstream branch), but broadcasts watermarks downstream (to all branches).For docs on routing/watermarks, please see https://getsentry.github.io/streams/runtime/arroyo.html#routes.