[PoC] Postgres implementation on transaction ID #22
Conversation
| @@ -73,15 +72,15 @@ func (s DefaultMySQLSchema) SelectQuery(topic string, consumerGroup string, offs | |||
| selectQuery := ` | |||
| SELECT offset, uuid, payload, metadata FROM ` + s.MessagesTable(topic) + ` | |||
| WHERE | |||
| offset > (` + nextOffsetQuery + `) | |||
| offset > (` + nextOffsetQuery + `) AND transaction_id < pg_snapshot_xmin(pg_current_snapshot()) | |||
There was a problem hiding this comment.
Hmm, looks unsuitable for MySQL (there is no snapshots)
There was a problem hiding this comment.
Stupid copy-paste 😅 Removed in this PR: #23
| (transaction_id > (SELECT last_processed_transaction_id FROM last_processed)) | ||
| ) | ||
| AND | ||
| transaction_id < pg_snapshot_xmin(pg_current_snapshot()) | ||
| ORDER BY | ||
| "offset" ASC |
There was a problem hiding this comment.
I suppose there should be
| "offset" ASC | |
| transaction_id ASC, | |
| "offset" ASC |
There was a problem hiding this comment.
Also, this type of ordering, should be covered by index like (transaction_id ASC, "offset" ASC). It's not mentioned in original article, but I faced performance issues when load testing my own implementation of adapter :)
There was a problem hiding this comment.
Good catch and point about the index!
I applied them in #23
| return msg.UUID, false, errors.Wrap(err, "cannot send consumed query") | ||
| msgCtx := setTxToContext(ctx, tx) | ||
|
|
||
| acked := s.sendMessage(msgCtx, msg, out, logger) |
There was a problem hiding this comment.
What do you think, if we send messages concurrently here? As I see the main problem here not in querying batch of messages, but in serial processing of the batch
There was a problem hiding this comment.
If I'm not missing anything, it would not be compatible with the current backing strategy 🤔 in other words - the current implementation would not work properly with handling out-of-order acks.
Like:
- Msg 1 is acked,
- Msg 2 is acked,
- Msg 3 is not acked
- Msg 4 is acked,
In theory, we could just ack the last message before nack 🤔 but it's a pretty specific scenario. So I think that it may be out of this PR scope. Batch querying already gives a lot of boost compared to querying message by message.
If you'll have some spare time and need such acking model, feel free to post PR with that! (As long as I understood you question properly)
No description provided.