Refactor state management out of BufferStrategy #13669
Conversation
cgardens
requested review from
tuliren and
ChristopheDuong
and removed request for
tuliren
| * | ||
| * THis install such a hook to be triggered when that happens. | ||
| */ | ||
| void registerFlushAllEventHook(VoidCallable onFlushAllEventHook); |
There was a problem hiding this comment.
this is the callback pattern that doesn't seem idiomatic. the fact that the event hook is set after the instantiation of object is a code smell.
There was a problem hiding this comment.
yes, the "flush all event" is indeed a code smell to be looked at, thank you for investigating this.
The buffering strategy refactor was already a pretty large PR on handling AirbyteMessage differently, handling the state messages in a good manner was not necessarily the primary priority there... So the hook "pattern" was introduced/hacked to avoid diverging too much from the previous behavior of destinations on that matter.
| @@ -91,22 +94,13 @@ public void flushAll() throws Exception { | |||
| }, Map.of("bufferSizeInBytes", bufferSizeInBytes)); | |||
| close(); | |||
| clear(); | |||
|
|
|||
| if (onFlushAllEventHook != null) { | |||
There was a problem hiding this comment.
the fact that all buffers need to replicate this behavior is a smell.
...main/java/io/airbyte/integrations/destination/record_buffer/SerializedBufferingStrategy.java
Show resolved
Hide resolved
cgardens
changed the title
...ava/io/airbyte/integrations/destination/buffered_stream_consumer/BufferedStreamConsumer.java
Outdated
Show resolved
Hide resolved
...ava/io/airbyte/integrations/destination/buffered_stream_consumer/BufferedStreamConsumer.java
Outdated
Show resolved
Hide resolved
...a/io/airbyte/integrations/destination/record_buffer/InMemoryRecordBufferingStrategyTest.java
Outdated
Show resolved
Hide resolved
…integrations/destination/buffered_stream_consumer/BufferedStreamConsumer.java Co-authored-by: Edward Gao <edward.gao@airbyte.io>
What
I was looking into how to update state handling in destinations to handle the new version of the protocol. In the meanwhile, I noticed that we had a weird pattern where the record buffers had a callback to handle tracking state messages. This is not idiomatic to java and mixes two different concerns.
Refactor the code to keep them separate.
See comment in line, but there's one thing in there that I want to make sure it isn't a bug.