Hi Fabian, Awesome, this project looks great! Thanks for sharing. It would work well with KafkaSource and the DataStream API as you've mentioned. We have something similar internally, but where we are encountering difficulty is integrating it with the Flink SQL Kafka DynamicTable Source and Sinks. Our Large Message SerDe uses the Kafka Message header to store the URI on object storage, and currently the Flink SQL Format Interfaces do not allow passing data to/from the Kafka message headers, which lead me to suggest my proposal. It's not easy for us to change our Large Message SerDe to use the value to provide the reference, as it's already widely used and would require a significant migration.
However, thinking further, maybe we should not bring in any Large message concerns into Flink, but instead better support reading and writing headers from Flink Formats. I'm aware of the existing work in progress on handling headers via FLIP-454 <https://cwiki.apache.org/confluence/display/FLINK/FLIP-454%3A+New+Apicurio+Avro+format> and this mailing list discussion <https://lists.apache.org/thread/f1y8dzplmsjdzcwsw60cml1rnfl57nol>. On Mon, Jul 8, 2024 at 10:08 AM Fabian Paul <fp...@apache.org> wrote: > Hi Kevin, > > I worked on a project [1] in the past that had a similar purpose. You > should be able to use a similar approach with the existing KafkaSource by > implementing your own KafkaRecordDeserializationSchema that hides the logic > of pulling the records from blob storage from the connector. You can even > use the linked project directly with the KafkaSource using [2] and [3]. > > I agree there is room for improvements, like propagating Flink's Filesystem > credentials to the custom deserializer, but the overall idea seems to > require only very few changes to Flink. > > Best, > Fabian > > [1] https://github.com/bakdata/kafka-large-message-serde > [2] > > https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/deserializer/KafkaRecordDeserializationSchema.java#L107 > [3] > > https://github.com/bakdata/kafka-large-message-serde/blob/09eae933afaf8a1970b1b1bebcdffe934c368cb9/large-message-serde/src/main/java/com/bakdata/kafka/LargeMessageDeserializer.java#L50 > > On Mon, Jul 8, 2024 at 3:49 PM Kevin Lam <kevin....@shopify.com.invalid> > wrote: > > > Hi all, > > > > Thanks for the responses. > > > > Grace those are indeed both challenges, thanks for flagging them. > Regarding > > expiry, we could consider having a Mark and Sweep garbage collection > > system. A service can consume the topics with large messages, and track > > references. When there are no references left for large messages, they > can > > be removed. > > > > Martjin, I will take a look at if there's any prior discussions in the > > Kafka community and send the proposal to the Kafka Dev mailing list if it > > makes sense :). It'd be much preferred if this was natively supported by > > Kafka, since it's not currently I was also exploring making this work in > > Flink. > > > > > > > > On Mon, Jul 8, 2024 at 3:23 AM Martijn Visser <martijnvis...@apache.org> > > wrote: > > > > > Hi Kevin, > > > > > > I just want to double check, were you planning to send this proposal to > > the > > > Kafka Dev mailing list? Because I don't see directly how this affects > > Flink > > > :) > > > > > > Best regards, > > > > > > Martijn > > > > > > On Mon, Jul 8, 2024 at 8:05 AM Grace Grimwood <ggrim...@redhat.com> > > wrote: > > > > > > > Hi Kevin, > > > > > > > > Thanks for starting this thread. > > > > > > > > This idea is something that was discussed in Kroxylicious (an open > > source > > > > Kafka proxy, I'm a maintainer there). In that discussion [1] we came > to > > > the > > > > conclusion that there are a couple of issues with implementing this: > > > > 1. Doesn't scale - very large messages (>1GiB) or large batch sizes > > could > > > > cause extreme memory bloat in clients, as the entire thing would need > > to > > > be > > > > fed into the producer which could very quickly fill its buffers. > > > Depending > > > > on how the subsequent deserialization and payload fetch is handled at > > the > > > > consumer end, it's likely that the same behaviour would also be seen > > > there. > > > > 2. Difficult to sync expiry - when Kafka deletes messages due to > > > retention > > > > (or topic compaction), it does so without notifying clients. There is > > no > > > > (easy) way to ensure the associated payload is deleted from object > > > storage > > > > at the same time. > > > > > > > > It's not totally clear how Conduktor solved these issues, but IMO > they > > > are > > > > worth keeping in mind. For Kroxylicious we decided these problems > meant > > > it > > > > wasn't practical for us to implement this, but I'd be curious to know > > if > > > > you've got any ideas :) > > > > > > > > Regards, > > > > Grace > > > > > > > > [1] https://github.com/kroxylicious/kroxylicious/discussions/1244 > > > > > > > > On Sat, Jul 6, 2024 at 8:21 AM Kevin Lam > <kevin....@shopify.com.invalid > > > > > > > wrote: > > > > > > > > > Hi all, > > > > > > > > > > Writing to see if the community would be open to exploring a FLIP > for > > > the > > > > > Kafka Table Connectors. The FLIP would allow for storing Kafka > > Messages > > > > > beyond a Kafka cluster's message limit (1 MB by default) out of > band > > in > > > > > cloud object storage or another backend. > > > > > > > > > > During serialization the message would be replaced with a > reference, > > > and > > > > > during deserialization the reference would be used to fetch the > large > > > > > message and pass it to Flink. Something like Option 1 in this blog > > post > > > > > < > > > > > > > > > > > > > > > https://www.conduktor.io/kafka/how-to-send-large-messages-in-apache-kafka/#Option-1:-using-an-external-store-(GB-size-messages)-0 > > > > > > > > > > > . > > > > > > > > > > What do you think? > > > > > > > > > > We can make it generic by allowing users to implement their own > > > > > LargeMessageSerializer/Deserializer interface for serializing and > > > > > deserializing and handling interactions with object storage or some > > > other > > > > > backend. > > > > > > > > > > The Kafka Connectors can be extended to support ConfigOptions to > > > > > specify the class to load, as well as some user-specified > properties. > > > For > > > > > example: `large-record-handling.class` and ` > > > > > large-record-handling.properties.*` (where the user can specify any > > > > > properties similar to how the Kafka Consumer and Producer > properties > > > are > > > > > handled > > > > > < > > > > > > > > > > > > > > > https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/KafkaDynamicTableFactory.java#L201 > > > > > > > > > > > ). > > > > > > > > > > In terms of call sites for the LargeMessage handling, I think we > can > > > > > consider inside of DynamicKafkaDeserializationSchema > > > > > < > > > > > > > > > > > > > > > https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaDeserializationSchema.java > > > > > > > > > > > and DynamicKafkaRecordSerializationSchema > > > > > < > > > > > > > > > > > > > > > https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java > > > > > >, > > > > > where the ConsumerRecord > > > > > < > > > > > > > > > > > > > > > https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaDeserializationSchema.java#L108 > > > > > > > > > > > and ProducerRecords > > > > > < > > > > > > > > > > > > > > > https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/streaming/connectors/kafka/table/DynamicKafkaRecordSerializationSchema.java#L75 > > > > > > > > > > > are passed respectively. > > > > > > > > > > If there's interest, I would be happy to help flesh out the > proposal > > > > more. > > > > > > > > > > > > > > >
