i guess this is related to an at-least-once vs at-most-once decision, but what would be the appropriate point in an onyx 0.10 plugin to consider a segment to be fully processed ? i’m not sure whether the synced? function or the checkpointed! function is the correct one to use, since both of them have access to the epoch

Synced means that it's safe for the barrier to proceed, checkpointed! means that the all the barriers for that epoch have been processed and the checkpoint has been written.

right, so checkpointed! is called before synced?

The main purpose of synced? is to give a synchronisation point at which the plugin can be sure things are written out to the next task or the output, especially for asynchronous actions, but if something fails before the final checkpoint has been communicated up to the coordinator and written, then you could end up with it all being performed again for that data.

After

Basically the coordinator waits for synced? = true on all tasks, then once the coordinator knows it’s synced? on all tasks, the coordinator communicates back down to tell the tasks to checkpointed! for that epoch.

ok

Writing a plugin?

how did you guess that 🙂

okay so then synced? would be the best place to mark things as fully processed for an input plugin

how do i know what the ‘next’ epoch will be after a synced ? is it always just incremented by 1 ?

Yes, but make sure you reset any state on a recover! call, since it means that it never moved to the next epoch.

And yes, next epoch is always incremented by 1, unless recover! is called, at which point it resets back to 0

ok

(and you get a new replica version passed in)

If you would like a code review at any point let me know.

Good to see how the plugin interface is being used.

thanks for that, i might actually make use of the offer 🙂

if you’re on board, i might do a PR of plugins/protocols.clj to document these semantics a bit better

The core async input plugin is kind of an ungodly hack, so try not to follow it too much.

Yes please!

ok will do

Thank you 🙂

one last question — what’s the appropriate way to tell poll! to back off for a while because there is no input now, but might be in a few seconds ? the ugly way would be to just call Thread/sleep for up to timeout-ms, but is there a better way (e.g. mark drained? )

Well, we pass in timeout-ms to you, I would just track timeout-ms and backoff with [java.util.concurrent.locks LockSupport] via LockSupport/parkNanos, and then our own poll! again

with a smaller timeout-ms

If it weren’t for this (when-not (nil? segment) https://github.com/onyx-platform/onyx/blob/0.10.x/src/onyx/peer/read_batch.clj#L40 it would track the poll time for you and then bail after batch-timeout.

Maybe it should work that way, I generally prefer to it to just get on with sending work down in that case though.

oh, that makes sense

The current behaviour is probably best for throughput, but it does come at the cost of some cpu burn.

yep

okay, thanks for the advice, will implement it as such

No worries