You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@flink.apache.org by Kevin Lam <ke...@shopify.com.INVALID> on 2024/07/05 20:13:37 UTC
Potential Kafka Connector FLIP: Large Message Handling
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.
Re: Potential Kafka Connector FLIP: Large Message Handling
Posted by Kevin Lam <ke...@shopify.com.INVALID>.
Apologies David, here's a temporary link to the image:
https://screenshot.click/19-19-0xwls-jfx7s.png
On Thu, Jul 18, 2024 at 9:00 AM David Radley <da...@uk.ibm.com>
wrote:
> Hi Kevin,
> That sounds good, unfortunately the image did not come through in the
> email for me, kind regards, David
>
> From: Kevin Lam <ke...@shopify.com.INVALID>
> Date: Wednesday, 10 July 2024 at 19:20
> To: dev@flink.apache.org <de...@flink.apache.org>
> Subject: [EXTERNAL] Re: Potential Kafka Connector FLIP: Large Message
> Handling
> Hey David, Yes my proposal was originally to do this at the connector
> level, as you mentioned it doesn't make sense in a format. In the end I
> took the approach in my previous e-mail: I was able to insert our Large
> Message handling by overriding
>
> Hey David,
>
> Yes my proposal was originally to do this at the connector level, as you
> mentioned it doesn't make sense in a format. In the end I took the approach
> in my previous e-mail:
>
> I was able to insert our Large Message handling by overriding
> value.serializer<
> https://kafka.apache.org/documentation/#producerconfigs_value.serializer>
> and value.deserializer<
> https://kafka.apache.org/documentation/#consumerconfigs_value.deserializer>
> in the consumer and producer configuration that Flink sets, using the
> `properties.*` option in the Kafka Connector. This approach doesn't require
> Flink to know anything about large messages or have any major changes made
> to it.
> Flink uses the ByteArray(De|S)erializers by default in its source<
> https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L457-L470>
> and sink<
> https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java#L82-L83>.
> Overriding the source serializer requires a small change to flip this
> boolean<
> https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L464>
> to make it overridable. I'm planning to start a separate thread to propose
> making `value.serializer` overridable.
>
> This is a rough representation of the before and after for producers
> (similar applies for consumers):
> [cid:ii_lyg5y5qx0]
> I assume you would want to add configuration to define where the external
> storage lives and authentication. Limitations around stack and heap sizes
> would be worth considering.
>
> Yes, we're implementing the 'claim check' pattern and this is definitely
> something we're considering! Thanks for raising it
>
> On Wed, Jul 10, 2024 at 11:06 AM David Radley <david_radley@uk.ibm.com
> <ma...@uk.ibm.com>> wrote:
> Hi Kevin,
> You mention the link
> https://www.conduktor.io/kafka/how-to-send-large-messages-in-apache-kafka/#Option-1:-using-an-external-store-(GB-size-messages)-0
> , I assume this is the approach you are considering. And that this is being
> done at the connector level, as the message could be in any of the existing
> supported formats – so is not appropriate as a new format. It sounds like
> for deserialization, the reference to external storage header would be
> found in your deser and the contents then taken from external source and
> put into the Kafka body or the other way round for serialization. This is
> different to my Apicurio work that is to handle format specific headers.
>
> I assume you would want to add configuration to define where the external
> storage lives and authentication. Limitations around stack and heap sizes
> would be worth considering.
>
> Am I understanding your intent correctly?
> Kind regards, David.
>
>
> From: Kevin Lam <ke...@shopify.com.INVALID>
> Date: Wednesday, 10 July 2024 at 14:35
> To: dev@flink.apache.org<ma...@flink.apache.org> <
> dev@flink.apache.org<ma...@flink.apache.org>>
> Subject: [EXTERNAL] Re: Potential Kafka Connector FLIP: Large Message
> Handling
> Hey all, just a follow-up here. I was able to insert our Large Message
> handling by overriding value.serializer
> <https://kafka.apache.org/documentation/#producerconfigs_value.serializer>
> and value.deserializer
> <
> https://kafka.apache.org/documentation/#consumerconfigs_value.deserializer
> >
> in the consumer and producer configuration that Flink sets, using the
> `properties.*` option in the Kafka Connector. This approach doesn't require
> Flink to know anything about large messages or have any major changes made
> to it.
>
> Flink uses the ByteArray(De|S)erializers by default in its source
> <
> https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L457-L470
> >
> and sink
> <
> https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java#L82-L83
> >.
> Overriding the source serializer requires a small change to flip this
> boolean
> <
> https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L464
> >
> to make it overridable. I'm planning to start a separate thread to propose
> making `value.serializer` overridable.
>
>
> On Mon, Jul 8, 2024 at 11:18 AM Kevin Lam <kevin.lam@shopify.com<mailto:
> kevin.lam@shopify.com>> wrote:
>
> > 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 <fpaul@apache.org<mailto:
> fpaul@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.lam@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 <
> martijnvisser@apache.org<ma...@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 <ggrimwoo@redhat.com
> <ma...@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.lam@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.
> >> > > > >
> >> > > >
> >> > >
> >> >
> >>
> >
>
> Unless otherwise stated above:
>
> IBM United Kingdom Limited
> Registered in England and Wales with number 741598
> Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU
>
> Unless otherwise stated above:
>
> IBM United Kingdom Limited
> Registered in England and Wales with number 741598
> Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU
>
RE: Potential Kafka Connector FLIP: Large Message Handling
Posted by David Radley <da...@uk.ibm.com>.
Hi Kevin,
That sounds good, unfortunately the image did not come through in the email for me, kind regards, David
From: Kevin Lam <ke...@shopify.com.INVALID>
Date: Wednesday, 10 July 2024 at 19:20
To: dev@flink.apache.org <de...@flink.apache.org>
Subject: [EXTERNAL] Re: Potential Kafka Connector FLIP: Large Message Handling
Hey David, Yes my proposal was originally to do this at the connector level, as you mentioned it doesn't make sense in a format. In the end I took the approach in my previous e-mail: I was able to insert our Large Message handling by overriding
Hey David,
Yes my proposal was originally to do this at the connector level, as you mentioned it doesn't make sense in a format. In the end I took the approach in my previous e-mail:
I was able to insert our Large Message handling by overriding value.serializer<https://kafka.apache.org/documentation/#producerconfigs_value.serializer> and value.deserializer<https://kafka.apache.org/documentation/#consumerconfigs_value.deserializer> in the consumer and producer configuration that Flink sets, using the `properties.*` option in the Kafka Connector. This approach doesn't require Flink to know anything about large messages or have any major changes made to it.
Flink uses the ByteArray(De|S)erializers by default in its source<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L457-L470> and sink<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java#L82-L83>. Overriding the source serializer requires a small change to flip this boolean<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L464> to make it overridable. I'm planning to start a separate thread to propose making `value.serializer` overridable.
This is a rough representation of the before and after for producers (similar applies for consumers):
[cid:ii_lyg5y5qx0]
I assume you would want to add configuration to define where the external storage lives and authentication. Limitations around stack and heap sizes would be worth considering.
Yes, we're implementing the 'claim check' pattern and this is definitely something we're considering! Thanks for raising it
On Wed, Jul 10, 2024 at 11:06 AM David Radley <da...@uk.ibm.com>> wrote:
Hi Kevin,
You mention the link https://www.conduktor.io/kafka/how-to-send-large-messages-in-apache-kafka/#Option-1:-using-an-external-store-(GB-size-messages)-0 , I assume this is the approach you are considering. And that this is being done at the connector level, as the message could be in any of the existing supported formats – so is not appropriate as a new format. It sounds like for deserialization, the reference to external storage header would be found in your deser and the contents then taken from external source and put into the Kafka body or the other way round for serialization. This is different to my Apicurio work that is to handle format specific headers.
I assume you would want to add configuration to define where the external storage lives and authentication. Limitations around stack and heap sizes would be worth considering.
Am I understanding your intent correctly?
Kind regards, David.
From: Kevin Lam <ke...@shopify.com.INVALID>
Date: Wednesday, 10 July 2024 at 14:35
To: dev@flink.apache.org<ma...@flink.apache.org> <de...@flink.apache.org>>
Subject: [EXTERNAL] Re: Potential Kafka Connector FLIP: Large Message Handling
Hey all, just a follow-up here. I was able to insert our Large Message
handling by overriding value.serializer
<https://kafka.apache.org/documentation/#producerconfigs_value.serializer>
and value.deserializer
<https://kafka.apache.org/documentation/#consumerconfigs_value.deserializer>
in the consumer and producer configuration that Flink sets, using the
`properties.*` option in the Kafka Connector. This approach doesn't require
Flink to know anything about large messages or have any major changes made
to it.
Flink uses the ByteArray(De|S)erializers by default in its source
<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L457-L470>
and sink
<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java#L82-L83>.
Overriding the source serializer requires a small change to flip this
boolean
<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L464>
to make it overridable. I'm planning to start a separate thread to propose
making `value.serializer` overridable.
On Mon, Jul 8, 2024 at 11:18 AM Kevin Lam <ke...@shopify.com>> wrote:
> 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 <ke...@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 <ma...@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 <gg...@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.lam@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.
>> > > > >
>> > > >
>> > >
>> >
>>
>
Unless otherwise stated above:
IBM United Kingdom Limited
Registered in England and Wales with number 741598
Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU
Unless otherwise stated above:
IBM United Kingdom Limited
Registered in England and Wales with number 741598
Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU
Re: Potential Kafka Connector FLIP: Large Message Handling
Posted by Kevin Lam <ke...@shopify.com.INVALID>.
Hey David,
Yes my proposal was originally to do this at the connector level, as you
mentioned it doesn't make sense in a format. In the end I took the approach
in my previous e-mail:
I was able to insert our Large Message handling by overriding
> value.serializer
> <https://kafka.apache.org/documentation/#producerconfigs_value.serializer>
> and value.deserializer
> <https://kafka.apache.org/documentation/#consumerconfigs_value.deserializer>
> in the consumer and producer configuration that Flink sets, using the
> `properties.*` option in the Kafka Connector. This approach doesn't require
> Flink to know anything about large messages or have any major changes made
> to it.
> Flink uses the ByteArray(De|S)erializers by default in its source
> <https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L457-L470>
> and sink
> <https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java#L82-L83>.
> Overriding the source serializer requires a small change to flip this
> boolean
> <https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L464> to
> make it overridable. I'm planning to start a separate thread to propose
> making `value.serializer` overridable.
This is a rough representation of the before and after for producers
(similar applies for consumers):
[image: image.png]
> I assume you would want to add configuration to define where the external
> storage lives and authentication. Limitations around stack and heap sizes
> would be worth considering.
Yes, we're implementing the 'claim check' pattern and this is definitely
something we're considering! Thanks for raising it
On Wed, Jul 10, 2024 at 11:06 AM David Radley <da...@uk.ibm.com>
wrote:
> Hi Kevin,
> You mention the link
> https://www.conduktor.io/kafka/how-to-send-large-messages-in-apache-kafka/#Option-1:-using-an-external-store-(GB-size-messages)-0
> , I assume this is the approach you are considering. And that this is being
> done at the connector level, as the message could be in any of the existing
> supported formats – so is not appropriate as a new format. It sounds like
> for deserialization, the reference to external storage header would be
> found in your deser and the contents then taken from external source and
> put into the Kafka body or the other way round for serialization. This is
> different to my Apicurio work that is to handle format specific headers.
>
> I assume you would want to add configuration to define where the external
> storage lives and authentication. Limitations around stack and heap sizes
> would be worth considering.
>
> Am I understanding your intent correctly?
> Kind regards, David.
>
>
> From: Kevin Lam <ke...@shopify.com.INVALID>
> Date: Wednesday, 10 July 2024 at 14:35
> To: dev@flink.apache.org <de...@flink.apache.org>
> Subject: [EXTERNAL] Re: Potential Kafka Connector FLIP: Large Message
> Handling
> Hey all, just a follow-up here. I was able to insert our Large Message
> handling by overriding value.serializer
> <https://kafka.apache.org/documentation/#producerconfigs_value.serializer>
> and value.deserializer
> <
> https://kafka.apache.org/documentation/#consumerconfigs_value.deserializer
> >
> in the consumer and producer configuration that Flink sets, using the
> `properties.*` option in the Kafka Connector. This approach doesn't require
> Flink to know anything about large messages or have any major changes made
> to it.
>
> Flink uses the ByteArray(De|S)erializers by default in its source
> <
> https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L457-L470
> >
> and sink
> <
> https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java#L82-L83
> >.
> Overriding the source serializer requires a small change to flip this
> boolean
> <
> https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L464
> >
> to make it overridable. I'm planning to start a separate thread to propose
> making `value.serializer` overridable.
>
>
> On Mon, Jul 8, 2024 at 11:18 AM Kevin Lam <ke...@shopify.com> wrote:
>
> > 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.lam@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 <
> martijnvisser@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 <gg...@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.lam@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.
> >> > > > >
> >> > > >
> >> > >
> >> >
> >>
> >
>
> Unless otherwise stated above:
>
> IBM United Kingdom Limited
> Registered in England and Wales with number 741598
> Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU
>
RE: Potential Kafka Connector FLIP: Large Message Handling
Posted by David Radley <da...@uk.ibm.com>.
Hi Kevin,
You mention the link https://www.conduktor.io/kafka/how-to-send-large-messages-in-apache-kafka/#Option-1:-using-an-external-store-(GB-size-messages)-0 , I assume this is the approach you are considering. And that this is being done at the connector level, as the message could be in any of the existing supported formats – so is not appropriate as a new format. It sounds like for deserialization, the reference to external storage header would be found in your deser and the contents then taken from external source and put into the Kafka body or the other way round for serialization. This is different to my Apicurio work that is to handle format specific headers.
I assume you would want to add configuration to define where the external storage lives and authentication. Limitations around stack and heap sizes would be worth considering.
Am I understanding your intent correctly?
Kind regards, David.
From: Kevin Lam <ke...@shopify.com.INVALID>
Date: Wednesday, 10 July 2024 at 14:35
To: dev@flink.apache.org <de...@flink.apache.org>
Subject: [EXTERNAL] Re: Potential Kafka Connector FLIP: Large Message Handling
Hey all, just a follow-up here. I was able to insert our Large Message
handling by overriding value.serializer
<https://kafka.apache.org/documentation/#producerconfigs_value.serializer>
and value.deserializer
<https://kafka.apache.org/documentation/#consumerconfigs_value.deserializer>
in the consumer and producer configuration that Flink sets, using the
`properties.*` option in the Kafka Connector. This approach doesn't require
Flink to know anything about large messages or have any major changes made
to it.
Flink uses the ByteArray(De|S)erializers by default in its source
<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L457-L470>
and sink
<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java#L82-L83>.
Overriding the source serializer requires a small change to flip this
boolean
<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L464>
to make it overridable. I'm planning to start a separate thread to propose
making `value.serializer` overridable.
On Mon, Jul 8, 2024 at 11:18 AM Kevin Lam <ke...@shopify.com> wrote:
> 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 <ke...@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 <martijnvisser@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 <gg...@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.lam@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.
>> > > > >
>> > > >
>> > >
>> >
>>
>
Unless otherwise stated above:
IBM United Kingdom Limited
Registered in England and Wales with number 741598
Registered office: PO Box 41, North Harbour, Portsmouth, Hants. PO6 3AU
Re: Potential Kafka Connector FLIP: Large Message Handling
Posted by Kevin Lam <ke...@shopify.com.INVALID>.
Hey all, just a follow-up here. I was able to insert our Large Message
handling by overriding value.serializer
<https://kafka.apache.org/documentation/#producerconfigs_value.serializer>
and value.deserializer
<https://kafka.apache.org/documentation/#consumerconfigs_value.deserializer>
in the consumer and producer configuration that Flink sets, using the
`properties.*` option in the Kafka Connector. This approach doesn't require
Flink to know anything about large messages or have any major changes made
to it.
Flink uses the ByteArray(De|S)erializers by default in its source
<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L457-L470>
and sink
<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/sink/KafkaSinkBuilder.java#L82-L83>.
Overriding the source serializer requires a small change to flip this
boolean
<https://github.com/apache/flink-connector-kafka/blob/15d3fbd4e65dae6c334e2386dd337d2bf423c216/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/KafkaSourceBuilder.java#L464>
to make it overridable. I'm planning to start a separate thread to propose
making `value.serializer` overridable.
On Mon, Jul 8, 2024 at 11:18 AM Kevin Lam <ke...@shopify.com> wrote:
> 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 <ke...@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 <martijnvisser@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 <gg...@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.lam@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.
>> > > > >
>> > > >
>> > >
>> >
>>
>
Re: Potential Kafka Connector FLIP: Large Message Handling
Posted by Kevin Lam <ke...@shopify.com.INVALID>.
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 <ke...@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 <ma...@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 <gg...@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.lam@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.
> > > > >
> > > >
> > >
> >
>
Re: Potential Kafka Connector FLIP: Large Message Handling
Posted by Fabian Paul <fp...@apache.org>.
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 <ke...@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 <ma...@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 <gg...@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.lam@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.
> > > >
> > >
> >
>
Re: Potential Kafka Connector FLIP: Large Message Handling
Posted by Kevin Lam <ke...@shopify.com.INVALID>.
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 <ma...@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 <gg...@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 <ke...@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.
> > >
> >
>
Re: Potential Kafka Connector FLIP: Large Message Handling
Posted by Martijn Visser <ma...@apache.org>.
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 <gg...@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 <ke...@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.
> >
>
Re: Potential Kafka Connector FLIP: Large Message Handling
Posted by Grace Grimwood <gg...@redhat.com>.
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 <ke...@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.
>