Implementing Finalize Backfill Hook for All Backfila Clients #358
Conversation
client-dynamodb-v2/src/main/kotlin/app/cash/backfila/client/dynamodbv2/DynamoDbBackfill.kt
Show resolved
Hide resolved
| // If multiple partitions finish at the same time they will retry due to the hibernate | ||
| // version mismatch on the DbBackfillRun. | ||
| val partitions = dbRunPartition.backfill_run.partitions( | ||
| session, | ||
| backfillRunner.factory.queryFactory, | ||
| ) | ||
| if (partitions.all { it.run_state == BackfillState.COMPLETE }) { | ||
| partitions |
There was a problem hiding this comment.
we should do these server side changes in a separate PR, and get them deployed before making the client side changes. Otherwise, a client developer might implement the finalize method when it doesn't do anything yet.
Also, there are problems with this code I mentioned in the previous PR. I will see if I can take a look at writing this correctly in the next few days
There was a problem hiding this comment.
Sounds good - let me know a commit when you're done and we can either integrate it with this PR or externally!
There was a problem hiding this comment.
Maybe we want a COMPLETE_&_FINALIZED state to make that clearer?
There was a problem hiding this comment.
@shellderp just curious if any progress was made here?
There was a problem hiding this comment.
I know how to code it up, read the version and retry both transactions if the 2nd read a different version. I can do this in a followup if you get the code merged as is (just add tests in this PR)
| when (e) { | ||
| is HttpException -> | ||
| when (e.code()) { | ||
| 404 -> logger.info { "No finalization endpoint found for ${backfillRunner.backfillName}, skipping" } |
There was a problem hiding this comment.
This should show an event in the backfill.
I wonder if we want something to indicate versions for a backfill. Like what Backfills have what capabilities.
There was a problem hiding this comment.
We might want an event even in the success case?
| @@ -183,14 +184,44 @@ class BatchAwaiter( | |||
| ), | |||
| ) | |||
|
|
|||
| // If all states are COMPLETE the whole backfill will be completed. | |||
| // If all states are COMPLETE, for every other partition, the whole backfill will be completed. | |||
There was a problem hiding this comment.
Should we have a state that is Finalizing?
There was a problem hiding this comment.
I don't think the extra complexity is needed
| @@ -13,3 +13,9 @@ data class PrepareBackfillConfig<Param : Any>( | |||
| val parameters: Param, | |||
| val dryRun: Boolean, | |||
| ) | |||
|
|
|||
| data class FinalizeBackfillConfig<Param : Any>( | |||
There was a problem hiding this comment.
I wonder if we need better names and/or structure for these.
There was a problem hiding this comment.
PartitionlessBackfillConfig? Or WholeBackfillConfig?
This is config that is meant to represent the whole run. WholeRunBackfillConfig? Also add the accessor above. And probably an accessor to Prepare here too.
| // If multiple partitions finish at the same time they will retry due to the hibernate | ||
| // version mismatch on the DbBackfillRun. | ||
| val partitions = dbRunPartition.backfill_run.partitions( | ||
| session, | ||
| backfillRunner.factory.queryFactory, | ||
| ) | ||
| if (partitions.all { it.run_state == BackfillState.COMPLETE }) { | ||
| partitions |
There was a problem hiding this comment.
Maybe we want a COMPLETE_&_FINALIZED state to make that clearer?
| @@ -10,4 +11,9 @@ abstract class FixedSetBackfill<Param : Any> : Backfill { | |||
| } | |||
|
|
|||
| abstract fun runOne(row: FixedSetRow, backfillConfig: BackfillConfig<Param>) | |||
|
|
|||
| /** | |||
| * Override this to do any work after the backfill completes. | |||
There was a problem hiding this comment.
FYI This is a test only implementation.
| val config = parametersOperator.constructBackfillConfig(request) | ||
| backfill.finalize(config) | ||
|
|
||
| return FinalizeBackfillResponse.Builder() |
There was a problem hiding this comment.
I like the // Return empty 200 to indicate it was successful similar comment above. Does it make sense to make that more apparant?
| @@ -48,4 +49,9 @@ abstract class S3DatasourceBackfill<R : Any, P : Any> : Backfill { | |||
| * Produces records from the S3 file. | |||
| */ | |||
| abstract val recordStrategy: RecordStrategy<R> | |||
|
|
|||
| /** | |||
| * Override this to do any work after the backfill completes. | |||
There was a problem hiding this comment.
I think we need to update the comment a bit:
Override this to do any final work after the backfill completes. Only one successful call is expected in your distributed system and this call must be idempotent.
Does that make sense?
| @@ -21,9 +21,9 @@ internal class BackfillRegisteredParameters @Inject constructor( | |||
| private val queryFactory: Query.Factory, | |||
| ) : HibernateBackfill<DbRegisteredBackfill, Id<DbRegisteredBackfill>, NoParameters>() { | |||
|
|
|||
| override fun runOne(id: Id<DbRegisteredBackfill>, config: BackfillConfig<NoParameters>) { | |||
| override fun runOne(pkey: Id<DbRegisteredBackfill>, config: BackfillConfig<NoParameters>) { | |||
There was a problem hiding this comment.
undo the changes to this class
| @@ -379,6 +393,18 @@ class BackfillRunner private constructor( | |||
| } | |||
| } | |||
|
|
|||
| suspend fun finalize(state: FinalizeState) { | |||
There was a problem hiding this comment.
@shellderp is this right wrt coroutines?
| @@ -168,10 +170,9 @@ class BatchAwaiter( | |||
| return backfillRunner.runBatchAsync(this, batch) | |||
| } | |||
|
|
|||
| private fun completePartition() { | |||
| private suspend fun completePartition() { | |||
There was a problem hiding this comment.
Also this, is it right wrt coroutines? @shellderp
No description provided.