mirror of
https://github.com/hanzoai/tasks.git
synced 2026-10-30 14:56:19 +00:00
Michael SnowdenandMind.R
4e1ebfdfea
<!-- Describe what has changed in this PR --> **What changed?** I defined the persistence.QueueV2 interface, which supersedes the persistence.Queue interface. I also added an implementation of this interface for Cassandra. The implementations for delete and merge are left out for now and will be implemented later. As a result, there is no logic to read/write from the queues table (to read/write ack levels). Later, the ReadMessages method will need to be updated to use this as the minimum message id. <!-- Tell your future self why have you made these changes --> **Why?** We plan on using this for the upcoming history task DLQ project. <!-- How have you verified this change? Tested locally? Added a unit test? Checked in staging env? --> **How did you test it?** The new code has 100% test coverage, even including a test for CAS application failures due to concurrent writers. I tested that particular one locally 100 times to verify that it wasn't flaky. <!-- Assuming the worst case, what can be broken when deploying this change to production? --> **Potential risks** The biggest risk IMO is that because we're doing this piecemeal, we may realize there's an issue in the database schema and then have to change it later, which will be a pain if we need to ensure it's backwards compatible, but we can prevent this by ensuring that those changes are not split across multiple releases. <!-- Is this PR a hotfix candidate or require that a notification be sent to the broader community? (Yes/No) --> **Is hotfix candidate?** No.
25 lines
681 B
Cassandra CQL
25 lines
681 B
Cassandra CQL
CREATE TABLE queues
|
|
(
|
|
queue_type int,
|
|
queue_name text,
|
|
metadata_payload blob,
|
|
metadata_encoding text,
|
|
version bigint,
|
|
PRIMARY KEY ((queue_type, queue_name))
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|
|
|
|
CREATE TABLE queue_messages
|
|
(
|
|
queue_type int,
|
|
queue_name text,
|
|
queue_partition int,
|
|
message_id bigint,
|
|
message_payload blob,
|
|
message_encoding text,
|
|
PRIMARY KEY ((queue_type, queue_name, queue_partition), message_id)
|
|
) WITH COMPACTION = {
|
|
'class': 'org.apache.cassandra.db.compaction.LeveledCompactionStrategy'
|
|
};
|