Flink exactly-once Reader and Writer #4
Conversation
|
Addresses comments that were left on the original pull request on |
| private volatile boolean running = true; | ||
|
|
||
| // checkpoint trigger callback, invoked when a checkpoint event is received. | ||
| // no need to be volatile, the source is driven by but one thread |
There was a problem hiding this comment.
text - driven by only one thread
There was a problem hiding this comment.
Fixing that. I thought that was valid English, though...
| try { | ||
| Exceptions.handleInterrupted( txn::abort ); | ||
| } catch (Exception e) { | ||
| suppressed = e; |
There was a problem hiding this comment.
There is actually a bug in my code that the suppressed exception is not always re-thrown. Fixing that.
Do you want me to log it anyways (that would double-log it, because suppressed exceptions are anyways printed in the stack trace of the parent exception).
There was a problem hiding this comment.
I thought not rethrowning was on purpose, that's why wanted it logged. No need to log now since exception is not lost.
| final Transaction<T> txn = this.currentTxn; | ||
| Preconditions.checkState(txn != null, "bug: no transaction object when performing state snapshot"); | ||
|
|
||
| if (log.isDebugEnabled()) { |
There was a problem hiding this comment.
we could remove this check
There was a problem hiding this comment.
Sure. I often do these guards when there are more than two arguments to the log statement parameters, because that involves packaging them as parameters into an array. The overhead is admittedly little, and readability is probably more important here.
| // ==> There should never be a case where we have no pending transaction here | ||
| // | ||
|
|
||
| if (txnsPendingCommit.isEmpty()) { |
There was a problem hiding this comment.
Preconditions.checkState
There was a problem hiding this comment.
yes, that's nicer
|
|
||
| // the big assumption is that this now actually works and that the transaction has not timed out, yet | ||
|
|
||
| // TODO: currently, if this fails, there is actually data loss |
There was a problem hiding this comment.
please create an issue for this and link it here, will be easier to track. This is the approach we take in pravega for TODOs
| // checkpoint can be null when restoring from a savepoint that | ||
| // did not include any state for that particular reader name | ||
| if (checkpoint != null) { | ||
| this.readerGroup.resetReadersToCheckpoint(checkpoint); |
There was a problem hiding this comment.
for the case where checkpoint is null (if failure happened even before the first checkpoint succeeded), the readergroup still needs to be reset to its initial position right? because readers might have already read some data which will be discarded, hence they have to reread them
There was a problem hiding this comment.
True, it is one of the followup questions that I also raised.
Unfortunately, we cannot do the call in that particular place, because if there was never a complete checkpoint, no checkpoint will be restored and the hooks will not be called.
There was a problem hiding this comment.
Please file an issue for this and we can address it as a separate task.
| // Setup utility. | ||
| private static final SetupUtils SETUP_UTILS = new SetupUtils(); | ||
|
|
||
| //Ensure each test completes within 30 seconds. |
There was a problem hiding this comment.
good catch, thanks!
|
|
||
| return (Checkpoint) SerializationUtils.deserialize(bytes); | ||
| } | ||
| } |
There was a problem hiding this comment.
Add newline for consistency, for other files too.
|
Addressed all comments except one, for which I would need some input: Is the only way to do this to create a new reader group each time a recovery happens? Or should we call 'readerOffline' with a zero position on each reader (currently the checkpoint coordinator has no view on how many readers exist - all pravega structures are sort of opaque to it). |
Moved code for the exactly-once reader and writer.
Fixes #3