Maintain parent/child shard ordering across shard splits/merges. #155
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Kinesis allows clients to rely on an invariant that, for a given partition key, the order of records added to the stream will be maintained. IE: given an input `pkey=x,val=1 pkey=x,val=2 pkey=x,val=3`, the values `1,2,3` will be seen in that order when processed by clients, so long as clients are careful. It does so by putting all records for a single partition key into a single shard, then maintaining ordering within that shard.
However, shards can be split and merge, to distribute load better and handle per-shard throughput limits. Kinesis does this currently by (one or many times) splitting a single shard into two or by merging two adjacent shards into one. When this occurs, Kinesis still allows for ordering consistency by detailing shard parent/child relationships within its `listShards` outputs. A split shard A will create children B and C, both with `ParentShardId=A`. A merging of shards A and B into C will create a new shard C with `ParentShardId=A,AdjacentParentShardId=B`. So long as clients fully process all records in parents (including adjacent parents) before processing the new shard, ordering will be maintained.
`kinesis-consumer` currently doesn't do this. Instead, upon the initial (and subsequent) `listShards` call, all visible shards immediately begin processing. Considering this case, where shards split, then merge, and each shard `X` contains a single record `rX`:
```
time ->
B
/ \
A D
\ /
C
```
record `rD` should be processed after both `rB` and `rC` are processed, and both `rB` and `rC` should wait for `rA` to be processed. By starting goroutines immediately, any ordering of `{rA,rB,rC,rD}` might occur within the original code.
This PR utilizes the `AllGroup` as a book-keeper of fully processed shards, with the `Consumer` calling `CloseShard` once it has finished a shard. `AllGroup` doesn't release a shard for processing until its parents have fully been processed, and the consumer just processes the shards it receives as it used to.
This PR created a new `CloseableGroup` interface rather than append to the existing `Group` interface to maintain backwards compatibility in existing code that may already implement the `Group` interface elsewhere. Different `Group` implementations don't get the ordering described above, but the default `Consumer` does.
|
Nice work thanks for the PR! |
mskonovalov
reviewed
Sep 15, 2024
| @@ -138,6 +144,10 @@ func (c *Consumer) Scan(ctx context.Context, fn ScanFunc) error { | |||
| return <-errc | |||
| } | |||
|
|
|||
| func (c *Consumer) scanSingleShard(ctx context.Context, shardID string, fn ScanFunc) error { | |||
There was a problem hiding this comment.
seems this function is not in use
mskonovalov
reviewed
Sep 15, 2024
| // given partition key's data will be provided to clients in-order, | ||
| // but when splits or joins happen, we need to process all parents prior | ||
| // to processing children or that ordering guarantee is not maintained. | ||
| if waitForCloseChannel(ctx, parent1) && waitForCloseChannel(ctx, parent2) { |
There was a problem hiding this comment.
here is a bug: even if both parent1 and parent2 are nil, we still send the shard to shagrdc and it starts a new go-routne ti scan the shard that is already is being scanned
This was referenced Sep 16, 2024
harlow
pushed a commit
that referenced
this pull request
Sep 16, 2024
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Kinesis allows clients to rely on an invariant that, for a given partition key, the order of records added to the stream will be maintained. IE: given an input
pkey=x,val=1 pkey=x,val=2 pkey=x,val=3, the values1,2,3will be seen in that order when processed by clients, so long as clients are careful. It does so by putting all records for a single partition key into a single shard, then maintaining ordering within that shard.However, shards can be split and merge, to distribute load better and handle per-shard throughput limits. Kinesis does this currently by (one or many times) splitting a single shard into two or by merging two adjacent shards into one. When this occurs, Kinesis still allows for ordering consistency by detailing shard parent/child relationships within its
listShardsoutputs. A split shard A will create children B and C, both withParentShardId=A. A merging of shards A and B into C will create a new shard C withParentShardId=A,AdjacentParentShardId=B. So long as clients fully process all records in parents (including adjacent parents) before processing the new shard, ordering will be maintained.kinesis-consumercurrently doesn't do this. Instead, upon the initial (and subsequent)listShardscall, all visible shards immediately begin processing. Considering this case, where shards split, then merge, and each shardXcontains a single recordrX:record
rDshould be processed after bothrBandrCare processed, and bothrBandrCshould wait forrAto be processed. By starting goroutines immediately, any ordering of{rA,rB,rC,rD}might occur within the original code.This PR utilizes the
AllGroupas a book-keeper of fully processed shards, with theConsumercallingCloseShardonce it has finished a shard.AllGroupdoesn't release a shard for processing until its parents have fully been processed, and the consumer just processes the shards it receives as it used to.This PR created a new
CloseableGroupinterface rather than append to the existingGroupinterface to maintain backwards compatibility in existing code that may already implement theGroupinterface elsewhere. DifferentGroupimplementations don't get the ordering described above, but the defaultConsumerdoes.