Bulk Load CDK Stream Incomplete Prefactor 2: Drop Queue Reader and Si…

[johnny-schmidt]

What

• MessageQueueReader is factored away completely
• MessageQueue interface simplified from <KEY, MESSAGE> to <MESSAGE>. (This is because the next refactor will want an unkeyed queue, also it's just cleaner.)
• SizedMessageQueue dropped, as well as the inner QueueChannel; now there's just a simple default abstract ChannelMessageQueue class wrapping a kotlin Channel
• The KEY part of the interface is now in a MessageQueueSupplier<KEY, MESSAGE>, with a default supplier being wrapped records by stream descriptor
• Memory handling taken out of the queue entirely; the default queue has a message type of Reserved<DestinationRecordWrapped>>
• Slight refactor on the last MemoryManager PR to make the Reservation into a Reserved wrapper for an associated type
• Reserved is now acquired by MessageQueueWriter and released by SpillToDiskTask at the end of a use block

Also:

• test changes for the new code/behaviors

Note

• Some of the test changes is just throwing mocks away -- the interface around the message queue got so thin that there was no need for them anymore
• The QueueWriter reserves memory for the queue at the beginning, but never releases it. That's fine for now -- the next refactor will make it elegant for it to do. all its work inside a use block
• Obviously it would be better to take the reservation in the InputConsumer, not the Writer -- again, the next refactor is going to combine them in a way that will make that convenient

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

1 Skipped Deployment

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	⬜️ Ignored (Inspect)	Visit Preview		Oct 5, 2024 11:44pm

[edgao]

nit: can this just be a type alias? https://kotlinlang.org/docs/type-aliases.html

[johnny-schmidt]

I think no? Because I need to instantiate it when initializing the QueueSupplier?