refactor onStart and onFinish to take runnables and executed them guarded by state #40855
Conversation
hendrikmuhs
added
>non-issue
:StorageEngine/Rollup
|
Pinging @elastic/es-analytics-geo |
|
Pinging @elastic/ml-core |
hendrikmuhs
changed the title
jimczi
left a comment
jimczi
left a comment
There was a problem hiding this comment.
Thanks @hendrikmuhs , I left some comments
| * @param next Runnable for the next phase | ||
| */ | ||
| protected abstract void onStartJob(long now); | ||
| protected abstract void onStart(long now, Runnable next); |
There was a problem hiding this comment.
Since we're expecting async calls to be fired in onStart we should provide an ActionListener<Void> rather than a Runnable ? This way the implementation can fail the job nicely if an error occurs ?
| stats.markStartSearch(); | ||
| doNextSearch(buildSearchRequest(), ActionListener.wrap(this::onSearchResponse, this::finishWithSearchFailure)); | ||
| }); | ||
| } catch (Exception e) { |
There was a problem hiding this comment.
I don't think we should catch the exception here, if an error occurs in onStart we should provide an ActionListener to let the caller deals with the error (e.g. call listener.onFailure(e)):
onStart(now, ActionListener.wrap((o) -> {
stats.markStartSearch();
doNextSearch(buildSearchRequest(), ActionListener.wrap(this::onSearchResponse, this::finishWithSearchFailure));
}, (e) -> doSaveState(finishAndSetState(), position.get(), () -> onFailure(e));
There was a problem hiding this comment.
I also wonder if we should call onFailure before we save the state to be consistent with onFinish : onFailure(Exception e, Runnable finishAndSaveState) ?
| // being persisted as STARTED but then stop the job | ||
| doSaveState(finishAndSetState(), position.get(), () -> {}); | ||
| }); | ||
| } catch (Exception e) { |
There was a problem hiding this comment.
I think we should rather document the fact that the Runnable provided with onFinish (and maybe onFailure) should always run (even on errors). If we don't expect async calls in onFinish we can also keep the current code and just switch the execution order of onFinish and doSaveState.
jimczi
left a comment
jimczi
left a comment
There was a problem hiding this comment.
I left one comment regarding the call to onFailure if onFinish throws an error but other than that the refactor looks good to me.
| // execute finishing tasks | ||
| onFinish(ActionListener.wrap( | ||
| r -> doSaveState(finishAndSetState(), position.get(), () -> {}), | ||
| e -> doSaveState(finishAndSetState(), position.get(), () -> onFailure(e)))); |
There was a problem hiding this comment.
since we already called onFinish I don't think we should call onFailure if saving the state throws an error. It's not ideal but I'd prefer that we ignore the error and just finish the job.
jimczi
left a comment
jimczi
left a comment
There was a problem hiding this comment.
LGTM, thanks for iterating on this !
…rded by state (#40855) refactor onStart and onFinish to take action listeners and execute them when indexer is in indexing state.
…rded by state (elastic#40855) refactor onStart and onFinish to take action listeners and execute them when indexer is in indexing state.
5 participants
refactor onStart and onFinish to take action listeners and execute them when indexer is in indexing state.
Motivation of this change: DataFrameTransforms requires a hooks to prepare inner state before the 1st search request and to change inner state before the indexer finishes.
Rollup used the hooks mostly for logging purposes ans should be unaffected by this change.