Add its own commit loop which resolves partition offsets conflicts

[mosyp]

Fixes #25

Here's log from tests right after the new consumer is added to the group:

2021-06-10T16:36:16 [INFO] - WORKER-1: Delta write for version 5 has completed in 602071 microseconds
2021-06-10T16:36:16 [INFO] - WORKER-2: Starting run loop.
2021-06-10T16:36:17 [INFO] - WORKER-1: Delta write for version 6 has completed in 516716 microseconds
2021-06-10T16:36:17 [INFO] - WORKER-1: REBALANCE - Received new partition assignment list [1, 0]
2021-06-10T16:36:17 [INFO] - WORKER-2: REBALANCE - Received new partition assignment list [2, 3]
2021-06-10T16:36:17 [INFO] - WORKER-2: Resetting state with partitions: [3, 2]
2021-06-10T16:36:17 [INFO] - WORKER-2: Seeking consumer to 3:9
2021-06-10T16:36:17 [INFO] - WORKER-2: Seeking consumer to 2:10
2021-06-10T16:36:17 [INFO] - WORKER-1: Resetting state with partitions: [1, 0]
2021-06-10T16:36:17 [INFO] - WORKER-1: Seeking consumer to 1:12
2021-06-10T16:36:17 [INFO] - WORKER-1: Seeking consumer to 0:25
2021-06-10T16:36:17 [INFO] - WORKER-2: Conflict offset for partition 3: state=8, delta=10
2021-06-10T16:36:17 [INFO] - WORKER-2: Conflict offset for partition 2: state=9, delta=11
2021-06-10T16:36:17 [INFO] - WORKER-2: Resetting state with partitions: [3, 2]
2021-06-10T16:36:17 [INFO] - WORKER-2: Seeking consumer to 3:11
2021-06-10T16:36:17 [INFO] - WORKER-2: Seeking consumer to 2:12
2021-06-10T16:36:18 [INFO] - WORKER-1: Delta write for version 7 has completed in 601173 microseconds
2021-06-10T16:36:19 [INFO] - WORKER-1: Delta write for version 8 has completed in 718753 microseconds
2021-06-10T16:36:19 [INFO] - WORKER-1: Delta write for version 9 has completed in 653510 microseconds
2021-06-10T16:36:20 [INFO] - WORKER-2: Delta write for version 10 has completed in 2585776 microseconds
2021-06-10T16:36:20 [INFO] - WORKER-1: Delta write for version 11 has completed in 1145232 microseconds
2021-06-10T16:36:21 [INFO] - WORKER-2: Delta write for version 12 has completed in 873619 microseconds

So what do we see here. Before, the WORKER-1 had all 4 partition, after introducing the WORKER-2, we've got first to have 0 and 1, second 2 and 3.

On rebalance, we clear all pending state and seek the consumers to the offsets from delta.

However you might notice that WORKER-2 is seeked twice. That's due to the distributive nature, first seek it to 3:9,2:10, because delta log versions are 3:8,2:9 (notice +1, because we seek to the next message). But then when WORKER-2 tries to do a commit it founds that the delta store is actually 3:10,2:11 now. That is because of WORKER-1 committed version 6 right in between/before rebalance (note that we first load a table and then subscribe a consumer, so there's a time gap).
But that's okay since WORKER-2 notices that (e.g. updated delta offsets) and seek its consumer to 3:11,2:12.
Right after that, WORKER-1 cannot write data to partitions 2,3 so no more conflicts are possible and hence both WORKER-1 and WORKER-2 are making commits to the delta without any conflicts or any other issues in parallel

[houqp]

looks good to me overall 👍

[houqp]

i feel like this unwrap and clone can be avoided if we refactor how we check and reset partition assignments. but since this is not in the hot path, the overhead should be minor, so we can do the clean in future PRs.

[mosyp]

Okay so this took me a while. Yes I don't need clone, but I couldn't make it compile otherwise.

The errors I was getting is something like cannot borrow as the immutable because it's already borrowed a mutable

[houqp]

yeah, to workaround that, we need to combine the rebalance check and partition_assignment.reset_with into a single method.