V3 streaming - #664
V3 streaming#664xeus2001 wants to merge 16 commits into
Conversation
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
…r numeric identifiers. Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
…t, default is normalized. Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Code Coverage
|
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Code Coverage
|
hirenkp2000
left a comment
hirenkp2000
left a comment
There was a problem hiding this comment.
All reviewed except:
- lib-base
- lib-jbon
Other than recovery and acknowledgement related comments, all comments are not-that important to start with first implementation and adjust accordingly.
| * When the _recoveryRequest_ is then stored to resume the operation, it will start at chunk `1`. This will then read chunk `2`, `3`, and `4` again, of which `3` and `4` have been written already! This is the responsibility of the consumer to either skip writes `3` or `4`, or to ignore the write errors due to the fact that these features do exist already. | ||
| * @since 3.0 | ||
| */ | ||
| abstract val recoveryRequest: StreamRequest |
There was a problem hiding this comment.
W.r.t. recovery, I have a different opinion. I believe, creating another Stream using IStreamSession.read with a recoveryRequest is better option than enforcing recovery on existing Stream. There can be situation where Stream is broken/disconnected/crashed for some reason. With current way of relying recovery via existing broken Stream can lead to unwanted complications at the implementation side in reusing same Stream. Alternatively, think, how we will recover, if entire app is crashed?
W.r.t. the documentation provided here, it is still useful. If we choose to remove recoveryRequest from here (which is what I suggest), we should still retain the documentation, may be at Stream level
There was a problem hiding this comment.
It was thought to be used exactly like you describe, but I agree that the documentation was unclear on it. This was now improved. Methods were now designed in a way that makes it hard to misuse and an example code was added into Stream to explain how the reading and writing generally should happen.
| * @since 3.0 | ||
| * @throws naksha.base.NakshaException with error [ILLEGAL_STATE][naksha.base.NakshaError.ILLEGAL_STATE] if the chunk was already closed. | ||
| */ | ||
| override fun close() { |
There was a problem hiding this comment.
Assuming chunk "closure" as a "successful persistance" has a risk similar to setting "autoCommit" as true.
I think, we also need to consider negative scenario. where chunk persistance may fail. And so allowing a mechanism for implementation to explicitly abort the operation OR accept partial failure (i.e. still continue with other chunks).
My thoughts are:
- alongwith
acknowledgewe also needfailurefor a Chunk at Stream level - Assuming autoclosing as a success, can lead to unintentional mistakes. I would prefer explicit calls to
acknowledgeorfailureinstead.
It need not be exactly like I suggest, but something similar where it is explicit feedback.
There was a problem hiding this comment.
This was just addressed, please review the changed design.
| * | ||
| * Therefore, it can happen that the recovery point is behind the current write _HEAD_. For example, assume a stream starts at version `0`, and chunks `1`, `2`, `3`, and `4` are read and written in parallel in four worker threads, because they do not intersect. Now worker #1 acknowledges chunk `1`, this will update the _recoveryRequest_ and move it to `1`. Assume worker `3` and `4` next confirm their writes. This will **not** update _recoveryRequest_, because the confirmation for chuk `2` is outstanding. Now assume, the read aborts or the write fails. | ||
| * | ||
| * When the _recoveryRequest_ is then stored to resume the operation, it will start at chunk `1`. This will then read chunk `2`, `3`, and `4` again, of which `3` and `4` have been written already! This is the responsibility of the consumer to either skip writes `3` or `4`, or to ignore the write errors due to the fact that these features do exist already. |
There was a problem hiding this comment.
How about allowing excludeVersions[] in the StreamRequest ?
This can be populated in the recoveryRequest (e.g. excludeVersions = [3,4]), which can indirectly help Stream implementations to ignore those versions during recovery, simplifying the process a bit without having to persist what got processed / what not.
There was a problem hiding this comment.
Yes, maybe the storage can do this. As the storage provides a serializable StreamRequest in the exception, it can embed such information. However, there is no general way how to describe how to recover and skip certain chunks as there is no standarized way how chunks are assembled. For a transaction the above works, but especially when the storage reorders features this is not the case, so we better prepare the writer for duplicates, when recovering.
| * @since 3.0 | ||
| */ | ||
| @get:JvmName("version") | ||
| val version: Long, |
There was a problem hiding this comment.
"version of transaction" seems a bit confusing. Features can have versions, but not Transactions.
Did we mean transaction Id itself? Else, try clarifying what is expected here as version.
There was a problem hiding this comment.
A transaction produces a version, which is why the version is mandatory and the transaction information are optional. In other words, the state switch between version A and B must be applied atomic, and all features being part of this state switch aka transaction, must have the same version.
Additionally this builds up a logical versioning inside of each feature. This is very same in IML and Naksha.
There was a problem hiding this comment.
I added id (good hint) and tried to improve the documentation to explain what version actually means and why it is there.
There was a problem hiding this comment.
Is it intentional to continue with "version" name for "id" field ?
@get:JvmName("version")
val id: Id?
| * @since 3.0 | ||
| */ | ||
| features: Array<StreamFeature> | ||
| ): StreamChunk(stream, features) No newline at end of file |
There was a problem hiding this comment.
we need similar mechanism as per comment on StreamChunk , as to, how to register acknowledgement or failure for StreamTransaction?
There was a problem hiding this comment.
Review again, I hope the mechanics are now clear.
There was a problem hiding this comment.
I reviewed. Added comments on new changes for acknowledgement and failure construct.
Once that is aligned for StreamChunk, then same will be automatically applicable for StreamTransaction as well. And this comment then can be considered as resolved.
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
…transaction. Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
…r CLI progress bar. Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
Signed-off-by: Alexander Lowey-Weber <alexander.weber@here.com>
hirenkp2000
left a comment
hirenkp2000
left a comment
There was a problem hiding this comment.
Apart from other review comments, build pipeline is failing and needs addressing.
| * If this chunk is closed. | ||
| * The amount of outstanding acknowledgements; defaults to `0`. | ||
| * | ||
| * The stream reading thread should add one for every processor and write that receives this chunk. |
There was a problem hiding this comment.
Purpose of acknowledgeCount is not clear.
Chunk is either written or not written, i.e. boolean result. So, purpose of counter is not clear.
| val acknowledgeCount: AtomicInt = AtomicInt(0) | ||
| ) { | ||
| /** | ||
| * A thread safe failure reporting, called e.g. by [IStreamSession.write][naksha.model.IStreamSession.write] to report a failed write. |
There was a problem hiding this comment.
For retry handling, I think, we are mixing roles between reader and writer.
Reader's job should be to fetch and return newer transaction / chunk.
Writer's job should be to write into target storage.
When there is write-error situation, Writer is the right component to handle retry, depending on whether it is retry-able and with what strategy (instant, fixed interval, exponential back-off etc). Alternatively, if not writer, than may be the choreographer which triggers the writer operation is better placed.
My correction would be to simply do stream.failed(chunk) (without retry)
| throw illegalState("Chunk already closed") | ||
| open fun acknowledge() { | ||
| while (true) { | ||
| val count = acknowledgeCount.get() |
There was a problem hiding this comment.
as per earlier comment, the purpose of acknowledgeCount is not clear.
My suggestion will be to simply do stream.acknowledge(chunk) when write operation is successful
| * @param minVersion the minimal version to read; defaults to `0`. | ||
| * @param ignoreTransactions if transactions can be ignored, even while the storage supports transactions, this allows certain optimizations to be performed, like reordering features and re-grouping to created filled chunks for faster writing and eventually faster copy; defaults to _false_. | ||
| * @param chunkSize the amount of features to pack into each [StreamChunk] chunk; only applies when the storage does not support history and transaction logs **or** _ignoreTransactions_ was explicitly set to _true_; defaults to `1000`. | ||
| * @param timeout the timeout duration to wait for [acknowledgement][Stream.acknowledge] and for the read storage; defaults to 5 minutes. |
There was a problem hiding this comment.
I have same comment as earlier, that read v/s write concerns are getting a bit mixed here.
Write timeout is appropriate to be handled at a Writer side.
As Reader is not responsible for retry, the write timeout is irrelevant input for Reader.
My suggestion is:
- read timeout is already managed at storage session level (i.e. newStreamSession)
- so, we don't need any timeout here in StreamRequest level
| * @since 3.0 | ||
| */ | ||
| @get:JvmName("version") | ||
| val version: Long, |
There was a problem hiding this comment.
Is it intentional to continue with "version" name for "id" field ?
@get:JvmName("version")
val id: Id?
| * @since 3.0 | ||
| */ | ||
| features: Array<StreamFeature> | ||
| ): StreamChunk(stream, features) No newline at end of file |
There was a problem hiding this comment.
I reviewed. Added comments on new changes for acknowledgement and failure construct.
Once that is aligned for StreamChunk, then same will be automatically applicable for StreamTransaction as well. And this comment then can be considered as resolved.
| while (true) { | ||
| val count = acknowledgeCount.get() | ||
| if (count <= 0) return | ||
| if (!acknowledgeCount.compareAndSet(count, count - 1)) continue |
There was a problem hiding this comment.
I think we will loop back on success compareAndSet so one acknowledge() invocation would drain all current acknowledgments.
| id, | ||
| Platform.copy(transaction, true), | ||
| features, | ||
| acknowledgeCount |
There was a problem hiding this comment.
Based on Stream documentation we need to copy chunk before returning it again for retry. In copy we are passing original AtomicInt into retry chunk, so we would just add to existing acknowledgeCount again I think.
| if (v is Id) return v | ||
| if (v !is String) return null | ||
| val id = Id(v) | ||
| if (update) map_set(platformObject(), id, v) |
There was a problem hiding this comment.
I think we should still have key as key here and id as value.
| * @see numberToText | ||
| */ | ||
| private fun _numberToText(number: Long): String { | ||
| if (number >= 0L) return numberToTextMap[number] ?: number.toString() |
There was a problem hiding this comment.
If we would have Id(x).text where x is already in numberToTextMap we would get for example for id(1L).text "naksha~transactions" but for Id("1").text "1" . Not sure if possible scenarios.
3 participants
No description provided.