Hi Flink Devs,
Thanks a lot for the valuable feedback from Arvid, Danny and David on the
design of this SQS Connector.
We had an in-depth discussion on Slack regarding the semantics and stages
of implementation. Below is the *conclusion* of our conversation.
- Currently, FLIP will concentrate on capturing the connector details,
design, and implementation specifically for AWS SQS Standard Queues with
at-least once semantics. Support for FIFO queues will be introduced later
as an extension to this FLIP.
- *Evaluation of Exactly Once Semantics for Standard Queues*
- Unfortunately, achieving exactly once semantics is not feasible for
Standard Queues due to the following reasons:
- There can be duplicate message deliveries with SQS standard
queues.
- A successful delete call does not guarantee the absolute
deletion of the messages.
Therefore, *we can only provide at-least-once delivery for Standard Queues.*
- *Evaluation of Exactly Once Semantics for FIFO Queues*
- Exactly once semantics is achievable for FIFO Queues. To accomplish
this, we need to keep track of seqNum (which is strictly increasing for
messages) and groupId. By using the seqNum, we can achieve exactly-once
semantics.
- *Manipulation of Invisible Messages*
- *For Standard Queues*:
- Since we are providing at-least-once delivery, there is no need
to manipulate invisible messages. After their visibility
timeout expires,
the messages are consumed again, and user code needs to
handle these corner
scenarios.
- *For FIFO Queues:*
- The challenge is that processing is halted if the previous
invisible messages are not deleted. Therefore, we need to
additionally keep
track of ReceiptHandle. In case messages remain in an
invisible state for
an extended period, we need to mark them visible again. This
can be quite
tricky, especially with multiple groupIds. The details and
design of this
mechanism will be an extension to this flip.
Thanks a lot for the feedback. We have updated the FLIP.
Updated Google Doc Link -
https://docs.google.com/document/d/1lreo27jNh0LkRs1Mj9B3wj3itrzMa38D4_XGryOIFks/edit?usp=sharing
Thanks
Saurabh & Abhi
-----------------------------------------------------------------------------------------
*Slack Conversation Details **snapshot* *<Feel Free to Skip>**: TLDR *
Saurabh SinghSaurabh Singh 6:06 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726576584259939>
Hi @Arvid Heise <https://apache-flink.slack.com/team/U07BQT3AA1W>Thank you
for your response. We believe that it might be more efficient to continue
this conversation via Slack DMs, and we can update the FLIP with our
conclusions afterwards. I hope this works for you, but please feel free to
let me know if you have any concerns.
Regarding the discussion points:
1. *SQS Limitation*: SQS has a limitation where you cannot perform
operations on invisible messages if their receipt handles (the reference to
the messages) are not available.
2. *SQS Message IDs*: Message IDs are unique random strings.
Now, we have two scenarios:
- *Messages become invisible but are not captured as part of the
state* (e.g.,
while reading): In this case, we can't do anything except wait for them to
become visible again, which is not ideal for FIFO queues.
- *Messages become invisible but are captured as part of the state* (e.g.,
failure to act on the NotifyCheckpointComplete call): Here, we would
merge all the message IDs read until the next completed checkpoint and then
delete them together.
Are you planning to store the message ids of all to-be-deleted messages in
the state? That certainly works for low-volume sources but ideally, you
would just store the highest message id and enumerate the other message ids
from it.
Yes, we need to store all message IDs individually because, in Standard SQS
queues, messages are delivered in an unordered manner.Initially, we
considered addressing both Standard and FIFO queues. However, after
discussions and further thought, it seems that the same approach may not be
ideal for FIFO queues. We’d like to seek your advice on limiting the scope
of this FLIP to focus exclusively on Standard queues for better
implementation and clarity. We believe this could help streamline the
process, and we could later have a follow-up discussion or extension to
explore adjustments and optimizations for FIFO queues.
What are your thoughts on this approach?
Arvid HeiseArvid Heise 7:07 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726580266021769>
It's completely fine to focus on one type of queue and add the other later
(even without FLIP). Of course, you should either fail on FIFO or document
that this works in a very limited way.
7:09 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726580384937009>
Arvid Heise
Storing all message IDs in the Flink state should be seen as a last resort.
It's okay but depending on the number of messages may blow up state size
significantly. It's much better than storing the payloads themselves (as we
need to do for deduplication operator).
Danny CranmerDanny Cranmer 7:10 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726580422350019>
was added to the conversation by Arvid Heise.
Arvid HeiseArvid Heise 7:11 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726580488752129>
It usually pays off to toss a few ideas back and forth though: maybe it's
enough to store the ID of the last committed message. If you have a way to
distinguish replayed and not replayed message (timestamp?), then you could
discard all replayed message until the ID pops up.
7:13 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726580596696739>
Arvid Heise
What else can you do with receipt handles? Does it make sense to store them
instead of the ID?
7:14 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726580691227369>
Arvid Heise
Also probably a stupid Q: what happens with already read message when I set
the timeout to 0? I'm assuming the respective timeout is attached to
message on read and thus changing the timeout later has no effect on them...
7:15 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726580751903719>
Arvid Heise
Lastly: you much contact do you have to the SQS folks? Could they add a way
to interact with invisible messages later? Then, we could focus on a simple
solution (possibly ignoring EOS) and add that later.
Wednesday, 18 September
Saurabh SinghSaurabh Singh 5:43 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726661587582269>
Thank you for your feedback. Please find our responses below:
1. Regarding message handling, we are indeed storing the receipt handles
for the messages in the state. This is necessary for performing the delete
operation on the messages later. In our earlier communication, we referred
to them as message IDs for simplicity and ease of understanding.
2. For *Standard SQS queues*, due to the lack of ordering guarantees, we
must store each receipt handle to ensure we can delete the messages
appropriately. The message IDs alone are not sufficient for this purpose.
3. Anytime we set the visibility timeout of a message to 0 (whether
immediately after reading or later), the message becomes visible again,
leading to reprocessing and duplication. Fine-tuning the visibility timeout
is key to avoiding this duplication. (Ref Link
<https://repost.aws/questions/QUnM3etA9fSDuMJngSvql5lw/sqs-visibility-timeout-0>
)
4. We are simply users of the AWS infrastructure, and do not have direct
interaction with the SQS development team. So it is not possible to
interact with them on this.
Let us know your thoughts on this? (edited)
[image: Amazon Web Services, Inc.]Amazon Web Services, Inc.
SQS Visibility Timeout 0
<https://repost.aws/questions/QUnM3etA9fSDuMJngSvql5lw/sqs-visibility-timeout-0>
I have an SQS queue with visibility timeout 0 to test how it works because
I intend to use that setting for something that is outside of scope of this
explanation.
I have a Lambda function trigger...
Thursday, 19 September
Danny CranmerDanny Cranmer 1:43 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726733595413099>
Disclaimer, I am not experienced with SQS.
Regarding message handling, we are indeed storing the receipt handles for
the messages in the state.
Is this code available somewhere for us to look at, might help with the
understanding.
we are indeed storing the receipt handles for the messages in the state
Are we also storing the actual messageID? We would likely need this in the
case when we need to deduplicate a message with a new read receiptI note
that standard queues only guarantee at-least-once delivery, so we cannot
really provide exactly-once in Flink without more extensive deduplication
logic
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues-at-least-once-delivery.html.
Unless the delete operation returns an error that we can retry, do you know
the behaviour in these conditions?Since we are storing the message ID
(receipt handles) in the state it sounds like the happy path is solid,
minus the above ^. For the "Messages become invisible but are not captured
as part of the state" condition I suppose we are also covered, but we will
introduce some latency increasing out-of-orderness.It sounds like we need
to track the receiptId and messageId in state until we have successfully
deleted in SQS. The receiptId is used to actually delete the message and the
messageId is needed to dedupe. I assume we will be controlling the
visibility period in the source? If so we can expose this to the user (with
sensible defaults) to tune, ofc it should be something > checkpoint
interval.
We are simply users of the AWS infrastructure, and do not have direct
interaction with the SQS development team. So it is not possible to
interact with them on this.
I know a few people that work at SQS, no promises but I have pinged them
and might be able to get some feedback from them (edited)
Arvid HeiseArvid Heise 2:37 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726736852216579>
I wonder if we get away with simply offering at least once and then expose
the message ID as metadata and use Flink to dedupe.
2:37 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726736863805459>
Arvid Heise
If I don't delete a message, are they stored indefintively?
2:40 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726737056514629>
Arvid Heise
Just to explain my line of thought: Storing per message data in the source
blows up state. We can do that for low volume sources or if we have complex
queries (e.g. joins which also stores per message data). If SQS is low
volume anyways, we might simply replay everything and dedupe via message
ID. That obviously will slow down recovery, so it's more likely to work if
messages have some inherent retention period.
Saurabh SinghSaurabh Singh 5:55 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726748737945309>
Thank you for taking time and your feedback:
Is this code available somewhere for us to look at? It might help with
understanding.
Currently, it is not available in any public domain. However, you can refer
to the documentation of messageId and ReceiptHandle here
<https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-queue-message-identifiers.html>
.
Are we also storing the actual messageID? We would likely need this in the
case when we need to deduplicate a message with a new read receipt.
At the moment, we are not storing the actual messageID. We only perform
delete operations, which require only the receipt handle. While storing the
messageID would indeed be useful for deduplication, it introduces the
additional overhead of retaining the messageID until its retention period
expires. Please note that as per the reference
<https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues-at-least-once-delivery.html>
even a successful delete operation is not 100% guaranteed, as the copy of
the message might not be deleted on the server that is unavailable, and you
might receive that message copy again.
I assume we will be controlling the visibility period in the source? If so,
we can expose this to the user (with sensible defaults) to tune, of course,
it should be something > checkpoint interval.
While we can manage this via APIs, it should be part of the infrastructure
(SQS) provisioning and not something to be tuned on demand. We will provide
guidelines to tune these parameters (Visibility Timeout/Retention Period)
with respect to checkpoint duration/interval/timeout.
I wonder if we get away with simply offering at least once.
Yes, based on the details and discussion, this seems like a good starting
point.
Expose the message ID as metadata and use Flink to dedupe.
Could you please expand on this? As we understand, we would need to
maintain the messageID in a data structure until the retention time
configured on the queue.
If I don't delete a message, are they stored indefinitely?
No, messages are not stored indefinitely. They expire or are deleted
automatically after the message retention period of the queue.
[image: docs.aws.amazon.com]docs.aws.amazon.com
Amazon SQS queue and message identifiers - Amazon Simple Queue Service
<https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-queue-message-identifiers.html>
Learn about the unique identifiers assigned to messages in standard and
FIFO queues, their formats, and how to effectively manage and track
messages using these identifiers.
[image: docs.aws.amazon.com]docs.aws.amazon.com
Amazon SQS at-least-once delivery - Amazon Simple Queue Service
<https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues-at-least-once-delivery.html>
Learn about the mechanisms and considerations involved in ensuring message
delivery at least once.
Danny CranmerDanny Cranmer 6:08 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726749531070739>
Is this code available somewhere for us to look at? It might help with
understanding.
Currently, it is not available in any public domain. However, you can refer
to the documentation of messageId and ReceiptHandle here
<https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-queue-message-identifiers.html>
.
I meant your connector code, assuming it exists
Expose the message ID as metadata and use Flink to dedupe.
Could you please expand on this? As we understand, we would need to
maintain the messageID in a data structure until the retention time
configured on the queue.
We support at-least-once by default, but then provide a special
deserialisation schema interface that gives access to the SQS message ID.
Then a user can include this in their records for downstream
deduplication. Example here
<https://github.com/apache/flink-connector-aws/blob/main/flink-connector-aws/flink-connector-kinesis/src/main/java/org/apache/flink/streaming/connectors/kinesis/serialization/KinesisDeserializationSchema.java#L64>
for Kinesis, we give access to seqNum etc (edited)
6:10 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726749655044349>
Danny Cranmer
I assume we will be controlling the visibility period in the source? If so,
we can expose this to the user (with sensible defaults) to tune, of course,
it should be something > checkpoint interval.
While we can manage this via APIs, it should be part of the infrastructure
(SQS) provisioning and not something to be tuned on demand. We will provide
guidelines to tune these parameters (Visibility Timeout/Retention Period)
with respect to checkpoint duration/interval/timeout.
ok, it is a shame that users will need to tune their queue to work with
Flink, but sounds sensible given the limitations (edited)
Saurabh SinghSaurabh Singh 6:43 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726751603344809>
Thanks for the quick response and suggestions.
I meant your connector code, assuming it exists.
At the moment, we have a naive version implemented and used for our
internal application. Unfortunately, sharing the code in its current form
could have legal implications. Therefore, we have initiated this FLIP to
develop this for the open-source community.
[image: :+1:]1
Friday, 20 September
Danny CranmerDanny Cranmer 12:05 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726770912823589>
@David Green <https://apache-flink.slack.com/team/U07NKKNBG5P> has agreed
to help us out. He is a Principal Engineer on the SQS team
David GreenDavid Green 12:05 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726770941081139>
was added to the conversation by Danny Cranmer.
Danny CranmerDanny Cranmer 12:11 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726771289630659>
Hey @David Green <https://apache-flink.slack.com/team/U07NKKNBG5P>, thanks
for jumping in [image: :wave:]We are trying to design an SQS source for
Flink and achieve exactly once semantics. Flink has a built-in
checkpointing mechanism, sources typically commit their current read
position in the checkpoint and resume from there upon failure.For the SQS
source we are trying to avoid storing all messages IDs and read receipts in
the state to avoid huge state sizes. It would be helpful if you can review
this thread and see if we have made any wrong assumptions, or would suggest
any alternative solutions.
David GreenDavid Green 12:46 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726773418874009>
Hi all, will read the thread to catch up. Is there a design doc or is it
all in the chat?
Danny CranmerDanny Cranmer 12:49 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726773595097219>
Here is the design, although it is probably lacking details of what we are
discussing
https://cwiki.apache.org/confluence/display/FLINK/FLIP-477+Amazon+SQS+Source+Connector
[image: :+1:]1
David GreenDavid Green 1:07 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726774654899819>
Is there a definition of "exactly once"?
https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/guarantees/
e.g. does exactly once relate to distributed state, or local state?
[image: nightlies.apache.org]nightlies.apache.org
Fault Tolerance Guarantees
<https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/guarantees/>
Fault Tolerance Guarantees of Data Sources and Sinks # Flink’s fault
tolerance mechanism recovers programs in the presence of failures and
continues to execute them. Such failures include machine hardware failures,
network failures, transient program failures, etc.
Flink can guarantee exactly-once state updates to user-defined state only
when the source participates in the snapshotting mechanism. The following
table lists the state update guarantees of Flink coupled with the bundled
connectors.
David GreenDavid Green 1:13 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726774992609999>
SQS standard queues provide at-least-once delivery. In rare cases, messages
can be delivered more than once.
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues.html
This potential for duplicate deliveries with SQS standard queues will make
exactly once semantics impractical to implement in a FLINK source
connector. Unlike SQS standard queues, SQS FIFO ordering means that you
won't receive a duplicate delivery out of order.It might be useful to
declare the capabilities of the SQS source connector separately for SQS
FIFO and SQS standard queues.
[image: docs.aws.amazon.com]docs.aws.amazon.com
Amazon SQS standard queues - Amazon Simple Queue Service
<https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/standard-queues.html>
Learn about the basics of standard queues, including their features, usage,
and best practices for integrating them into your applications as part of a
messaging system.
1:18 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726775287577599>
David Green
For SQS FIFO queues, you could consider using the `SequenceNumber` provided
by [ReceiveMessage](
https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html
)
Tracking the `SequenceNumber` for each message group that is inflight might
be a good way to provide exactly once semantics in the FLINK source
connector.
[image: docs.aws.amazon.com]docs.aws.amazon.com
ReceiveMessage - Amazon Simple Queue Service
<https://docs.aws.amazon.com/AWSSimpleQueueService/latest/APIReference/API_ReceiveMessage.html>
Retrieves one or more messages (up to 10), from the specified queue. Using
the WaitTimeSeconds parameter enables long-poll support. For more
information, see Amazon SQS Long Polling in the Amazon SQS Developer Guide .
Danny CranmerDanny Cranmer 1:21 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726775472945699>
Exactly once means that each record is consumed and processed within the
Job only once. Even when there are failures and restarts. Flink has
"in-flight" state and checkpointed state. Upon failure Flink restores from
checkpointed state and discards "in-flight" state. Checkpoints are
consistent across parallel tasks of the Flink job
Danny CranmerDanny Cranmer 1:23 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726775590547299>
What is the sequence number? These docs are not that useful "Returns the
value provided by Amazon SQS."
1 reply
5 days agoView thread
David GreenDavid Green 1:23 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726775611304669>
Forgive my unfamiliarity with FLINK. What does that mean from a source
connector perspective? If a FLINK application crashes and is restarted on
another host, what is the expected behaviour?
Danny CranmerDanny Cranmer 1:28 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726775900380639>
The Flink job would resume from the last checkpoint. The source should then
resume from where it was at that checkpoint time/offset/etc. The idea for a
queue is that upon checkpoint completion we delete all the messages
consumed since the last checkpoint. However this cleanup is not guaranteed
to happen, so we often store metadata in the state to help us cleanup
eventually consistently and dedupe (edited)
1:32 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726776120746739>
Danny Cranmer
So one idea is to collect the receipt IDs and then delete the messages in
the checkpoint completion event. But since this might not happen we want to
store state to eventually cleanup without storing every message ID. Usually
we use some offset/sequence number.
1:33 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726776221119169>
Danny Cranmer
Additionally on failure, we might need to reread messages that were
in-flight, and we do not have the receipt IDs. But they may not be visible
now, so besides tuning the visibility period is there a better way to
optimise this?
Danny CranmerDanny Cranmer 1:38 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726776534516009>
Sorry for the vague questions, it's difficult when I do not know SQS and
you do not know Flink. Hopefully @Saurabh Singh
<https://apache-flink.slack.com/team/U06GJSZTEUC> will have some more
specific questions when online
[image: :+1:]1
David GreenDavid Green 1:59 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726777789811769>
Additionally on failure, we might need to reread messages that were
in-flight, and we do not have the receipt IDs. But they may not be visible
now, so besides tuning the visibility period is there a better way to
optimise this?
Unfortunately no - if a message was received and is still within its
invisibility timeout, there is no way to make it visible again without
using the ReceiptHandle.
2:01 <https://apache-flink.slack.com/archives/C07MRSRNWBU/p1726777889954539>
David Green
So one idea is to collect the receipt IDs and then delete the messages in
the checkpoint completion event. But since this might not happen we want to
store state to eventually cleanup without storing every message ID. Usually
we use some offset/sequence number.
This approach could work for SQS FIFO queues if you stored the
SequenceNumber by message group ID, but each message group with in-flight
messages would have to wait for those messages to become visible again.
Yesterday
Saurabh SinghSaurabh Singh 6:04 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1727094880363379>
Thanks @Danny Cranmer <https://apache-flink.slack.com/team/U03JD1ZFQ3D> @David
Green <https://apache-flink.slack.com/team/U07NKKNBG5P> for the valuable
insights. We have reviewed the conversation and are providing a summary
here. We will update the FLIP accordingly once we agree on the same.
- *Exactly Once Semantics for Standard Queues*
- Unfortunately, achieving exactly once semantics is not feasible for
Standard Queues due to the following reasons:
- There can be duplicate message deliveries with SQS standard queues.
- A successful delete call does not guarantee the absolute deletion
of the messages.
*Therefore, we can only provide at-least-once delivery for Standard
Queues.*• *Exactly Once Semantics for FIFO Queues*
- Exactly once semantics is achievable for FIFO Queues. To accomplish
this, we need to keep track of seqNum (which is strictly increasing for
messages) and groupId. By using the seqNum, we can achieve exactly once
semantics.• *Manipulation of Invisible Messages*
- *For Standard Queues:*
- Since we are providing at-least-once delivery, there is no need to
manipulate invisible messages. After their visibility timeout expires, the
messages are consumed again, and user code needs to handle these corner
scenarios.
-* For FIFO Queues:*
- The challenge is that processing is halted if the previous
invisible messages are not deleted. Therefore, we need to additionally keep
track of ReceiptHandle. In case messages remain in an invisible state for
an extended period, we need to mark them visible again. This can be quite
tricky, especially with multiple groupIds. I believe we need to study this
in detail and design the mechanism in the extension flip.Let us know if
this is good and we can proceed?
------------------------------
New
David GreenDavid Green 10:06 PM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1727109383466049>
LGTM
Today
Arvid HeiseArvid Heise 11:19 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1727156995566719>
Just to clarify:
- EOS with FIFO is out of scope of the FLIP now and there will be a
follow-up FLIP?
11:22
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1727157161772559>
Arvid Heise
Imho (and this view is not fully shared by all Flink committers) you don't
strictly need FLIPs for connectors, especially if you just submit
extensions. However, you can see how beneficial it is to discuss the ideas
with a broader audience.
Having a FLIP for the start of a connector is definitively something that
I'd like to see but we would be driven mad if there is FLIP for each and
every improvement. It's a different story if you change existing behavior.
Saurabh SinghSaurabh Singh 11:33 AM
<https://apache-flink.slack.com/archives/C07MRSRNWBU/p1727157813737819>
Agree @Arvid Heise <https://apache-flink.slack.com/team/U07BQT3AA1W>. We
will just extend the existing FLIP to document the behaviour FIFO queues.
This discussion has been very valuable for us to extend our knowledge and
understanding of the concepts and concluding to the right behaviour for the
connector. Thanks for helping us out here.
We will update the FLIP and discussion thread with the new details.
-------------------------------------------------------------------------------------------
On Tue, Sep 17, 2024 at 12:26 PM Arvid Heise <ar...@apache.org> wrote:
> Hi Saurabh,
>
> thank you very much for taking the time to explain SQS features to me.
> It's a cool concept that I haven't seen before.
>
> While your presented approach already works for Flink EOS, I still need to
> poke a bit because I think the usability may be improved.
>
> Some background information on how sinks usually work:
> * On checkpoint, we store the committables (=transactions or file
> locations).
> * On notifyCheckpointComplete, we commit them all or fail on persisting
> errors.
> * On recovery, we retry all stored committables.
> * On recovery, we cancel all opened transactions not belonging to the
> recovered state.
>
> Now back to SQS, can we speed up recovery? Let's go over my thoughts:
> * I'm assuming you need to set the timeout to a value that corresponds to
> Flink checkpoint duration. So I often see 1 minute checkpointing interval
> and a checkpoint taking some seconds, so we would choose a SQS timeout of 2
> minutes (we'd probably set it higher in realistic settings).
> * Before failure, Flink tries to read message1 and fails, there are more
> messages afterwards (message2, ...)
> * In standard queues, Flink restarts and will not see message1 until the
> timeout happens. It can immediately start processing message2 and so on.
> After 2 minutes, it will finally see message1 and consume it.
> * In FIFO queues, Flink will not see any message until the timeout
> happens. So effectively, Flink will not process anything for 2 minutes.
> That means that recovery is effectively always as high as the SQS timeout.
>
> I'm curious if there is a way to manually timeout the messages for faster
> recovery? I saw that you can decrease the timeout of specific messages,
> which could be used to set the timeout to 0 on restart of message1. If that
> works, we could do similar things to the sink. On restart, we immediately
> try to delete the messages that are part of the checkpoint and set the
> timeout to 0 for messages that were read after the checkpoint. If it's
> somehow possible to enumerate all messages before and past the respective
> checkpoint, we would not depend on the SQS timeout of the consumer for
> correctness and recovery time and could set it to its max=12h. Else we
> would have the users to trade off correctness (timeout too low) vs recovery
> time (timeout too high).
>
> What are you planning to store as the source state?
> > *record2 is still captured as part of the state, so it gets deleted
> from the queue during the next checkpoint.*
> Are you planning to store the message ids of all to-be-deleted messages in
> the state? That certainly works for low volume sources but ideally, you
> would just store the highest message id and enumerate the other message ids
> from it.
>
> Could you please update the FLIP to include the provided information so
> that other readers can quickly understand the visibility timeout and the
> state management? (external links are fine of course but please provide a
> brief summary in case the link becomes dead at some point) You can of
> course defer it until we are completely aligned.
>
> Best,
>
> Arvid
>
> On Fri, Sep 13, 2024 at 9:32 AM Saurabh Singh <saurabhsingh9...@gmail.com>
> wrote:
>
>> Hi Arvid,
>>
>> Thanks a lot for your review.
>>
>> 1. Exactly once:
>>
>>
>>
>> *When using an SQS source, we leverage two key properties of SQS:1.
>> Message Retention: This property ensures that messages are retained in the
>> queue for a specified period, allowing consumers enough time to process
>> them. Post these messages are deleted.2. Visibility Timeout [1] : This
>> property prevents other consumers from receiving and processing the same
>> message while it's being processed by the current consumer.*
>>
>> *Regarding the scenario described below:*
>>
>> * Read record2, emit record2, record2 is written into a Kafka transaction2
>> * Checkpoint2 happens, state is checkpointed, sink remembers transaction2
>> to be committed
>> * NotifyCheckpointCompleted2 is lost for some reason or happens after next
>> failure
>> * Some failure happens, Flink is restarted with the current system state:
>> SQS contains record2, record3, Kafka contains record1 and pending record2.
>> On restart, sink will commit transactions2 recovered from the state. No
>> SQS
>> contains record2, record3 and Kafka contains record1, record2.
>> * Read record2, emit record2, record2 is written into a Kafka transaction3
>> * Eventually, we end up with two record2 in the sink.
>>
>> *In this scenario, we set the Message Visibility Timeout to be multiples
>> of the checkpoint duration and ensure it is longer than the recovery
>> duration. This way, when recovery occurs, record2 remains invisible and is
>> not read again, preventing duplication. record2 is still captured as part
>> of the state, so it gets deleted from the queue during the next checkpoint.*
>>
>> *Message Retention Time: We typically set this to a sufficiently long
>> duration. This is beneficial for recovering from extended outages and
>> backfilling data. If a message is not deleted, it will be read again once
>> it becomes visible after the visibility timeout expires.*
>>
>> *This is our current understanding for achieving exactly-once processing.
>> Do you see any challenges with this approach? If so, we would greatly
>> appreciate your feedback.*
>>
>> 2. Deleting messages:
>> *Yes, It's quite natural for consumers to delete messages for the reasons
>> mentioned above, such as recovery from outages and backfilling data.*
>>
>> 3. Regarding the Source parallelism:
>> *ACK. We'll document this behavior accordingly.*
>>
>> 4. Multiple queue support:
>> *Yes, as an extension to this FLIP, we will support this functionality in
>> the future. However, in the meantime, this can still be achieved by
>> creating multiple sources with specific queues.*
>>
>>
>> Please review and let us know your feedback on this.
>>
>> [1]
>> https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-visibility-timeout.html
>>
>>
>> Thanks
>> Saurabh & Abhi
>>
>>
>> On Thu, Sep 12, 2024 at 2:23 PM Arvid Heise <ar...@apache.org> wrote:
>>
>>> <I accidentally sent this message Saurabh directly instead to the mailing
>>> list, please disregard that first message and only respond to this one>
>>>
>>> Hi Saurabh,
>>>
>>> thank you for addressing the points. I still have concerns around EOS
>>> and a
>>> few remarks.
>>>
>>> 1. Exactly once:
>>> The checkpoint subsuming contract typically only applies to sinks, so
>>> it's
>>> unusual to be used in sources. Let's quickly look at an example so we are
>>> all on the same page.
>>>
>>> Assume we read messages record1, record2, record3 all arriving a few
>>> minutes apart such there are checkpoints in-between. For simplicity let's
>>> assume one checkpoint between records and call them checkpoint1 after 1,
>>> checkpoint2 after 2 and so on. The data is simply stored in some EOS sink
>>> (e.g. Kafka). The corresponding calls of NotifyCheckpointCompleted are
>>> enumerated with NotifyCheckpointCompleted1, NotifyCheckpointCompleted2,
>>> ...
>>>
>>> Then your source sees the following on failure:
>>> * Read record1, emit record1, record1 is written into a Kafka
>>> transaction1
>>> * Checkpoint1 happens, state is checkpointed, sink remembers transaction1
>>> to be committed
>>> * NotifyCheckpointCompleted1 happens, sink commits transaction1, so
>>> record1
>>> is visible to downstream consumer
>>> * During NotifyCheckpointCompleted1, SQS source also deletes record1
>>> * Some failure happens, Flink is restarted with the current system state:
>>> SQS contains record2, record3, Kafka contains record1. Since transaction1
>>> is already committed, the sink ignores it. So far so good.
>>>
>>> * Read record2, emit record2, record2 is written into a Kafka
>>> transaction2
>>> * Checkpoint2 happens, state is checkpointed, sink remembers transaction2
>>> to be committed
>>> * NotifyCheckpointCompleted2 is lost for some reason or happens after
>>> next
>>> failure
>>> * Some failure happens, Flink is restarted with the current system state:
>>> SQS contains record2, record3, Kafka contains record1 and pending
>>> record2.
>>> On restart, sink will commit transactions2 recovered from the state. No
>>> SQS
>>> contains record2, record3 and Kafka contains record1, record2.
>>> * Read record2, emit record2, record2 is written into a Kafka
>>> transaction3
>>> * Eventually, we end up with two record2 in the sink.
>>>
>>> So you see that with your proposal, we get duplicate records easily on
>>> failure, which is not exactly once. It's not enough to delete the
>>> messages
>>> eventually with NotifyCheckpointCompleted3.
>>>
>>> How it usually works:
>>> * The checkpoint state of the source contains some kind of offset, which
>>> would point to record3 in our example.
>>> * On recovery, seek the record corresponding to the offset and start
>>> reading from there.
>>>
>>> If the source system doesn't support seeking, then the usual way is to
>>> replay a larger section but discard all records with a smaller offset. If
>>> there is no strict ordering, a last resort is to track all message ids on
>>> low volume sources and discard duplicates. You could also start with just
>>> offering AT LEAST ONCE and let the user deduplicate. We should just be
>>> wary
>>> to not call something exactly once that isn't.
>>>
>>> 2. Deleting messages:
>>> It's rather uncommon for a connector to explicitly delete messages.
>>> Usually, they fall off retention at some point. I'd try to decouple that
>>> from EOS and give users an option to fully consume or not. Unless of
>>> course, it's very natural for SQS consumers to delete the messages; I
>>> can't
>>> assess that as I have never used SQS.
>>>
>>> 3. Regarding the parallelism:
>>> > *We have reviewed the DynamicParallelismInference and noticed that it
>>> is
>>> currently limited to batching jobs only [2]. *
>>> You are correct it cannot be used and thus we cannot enforce parallelism
>>> of
>>> 1. I got excited when I saw that interface and didn't read properly.
>>>
>>> Limiting the parallelism will help to reduce the resources. If you spawn
>>> more than 1 source subtask, it should get immediately terminated because
>>> there is nothing to do (don't forget to emit SourceReaderFinishedEvent
>>> for
>>> all idle subtasks). If the pipeline is embarrassing parallel, all
>>> downstream subtasks will also be closed.
>>> But there is nothing that you can do except document it to set the
>>> parallelism of the source to 1. A higher parallelism is only usable with
>>> a
>>> shuffle then.
>>>
>>> Alternatively, you could think about allowing the source to subscribe to
>>> multiple queues and distribute them to the different subtasks.
>>>
>>> Best,
>>>
>>> Arvid
>>>
>>> On Tue, Sep 10, 2024 at 4:50 PM Saurabh Singh <
>>> saurabhsingh9...@gmail.com>
>>> wrote:
>>>
>>> > Hi Danny, Arvid,
>>> >
>>> > Thank you so much for reviewing the FLIP. We apologize for the delayed
>>> > response; we've been quite busy over the past week.
>>> >
>>> >
>>> > - // .setFailOnError(false)
>>> > A general callout that this is not ideal, I know we use it elsewhere
>>> > but we should consider some pluggable error handling at some point.
>>> Not as
>>> > part of this FLIP, but if we can avoid adding this flag here, then
>>> I would.
>>> >
>>> > *Acknowledged. We will remove this flag and address it separately to
>>> > enhance our error handling design.*
>>> >
>>> > - // The code registers a checkpoint completion notification
>>> callback
>>> > via *notifyCheckpointComplete*
>>> > Flink does not guarantee that notifications are actually invoked and
>>> > successful [1]. Therefore there is a chance this method will not
>>> run, and
>>> > hence we could violate exactly once semantics here. Suggest that we
>>> > instead/additionally track the SQS read messages within the
>>> checkpoint
>>> > state, we can evict from state once deleted, and possibly prune on
>>> startup
>>> > or skip duplicates. Thoughts?
>>> >
>>> > *Yes, our implementation will comply with the checkpoint subsuming
>>> > contract mentioned in [1]. To achieve this, we will track newly read
>>> > messages <message ids> in a structure associated with a checkpoint ID.
>>> In
>>> > cases where checkpoint notifications are missed, we will review all
>>> > messages in the state with checkpoint IDs less than the received one,
>>> > delete those messages, and remove them from the state.*
>>> >
>>> > *In addition, for optimal functionality, we depend on the visibility
>>> > timeout configured for the queue. If read messages are not deleted
>>> before
>>> > the visibility timeout expires due to any transient issues, they will
>>> > reappear. We will offer guidance to users of this connector on
>>> adjusting
>>> > the visibility timeout value to align with specific checkpoint
>>> durations
>>> > and avoid duplicate processing.*
>>> >
>>> > - Since we only have 1 split, we should also limit the parallelism
>>> of
>>> > the source accordingly. We could exploit the
>>> DynamicParallelismInference to
>>> > effectively limit it to 1 unless the user explicitly overwrites.
>>> >
>>> > *We have reviewed the DynamicParallelismInference and noticed that it
>>> is
>>> > currently limited to batching jobs only [2]. We would appreciate it if
>>> you
>>> > could help us understand the rationale behind restricting the
>>> parallelism
>>> > for the source to 1. Are there any benefits to this limitation, or
>>> > potential issues if we do not enforce it?*
>>> >
>>> > - If I haven't overlooked something obvious, then your exactly-once
>>> > strategy will not work. There is unfortunately no guarantee that
>>> > notifyCheckpointComplete is called at all before a failure happens.
>>> So
>>> > you very likely get duplicate messages. Scanning the SQS
>>> > documentation, I saw that you can read the SequenceNumber of the
>>> message.
>>> > If you also store the latest number in the checkpoint state of the
>>> split,
>>> > then you can discard all messages during recovery that are smaller.
>>> But I
>>> > have never used SQS, so just take it as an inspiration. Also note
>>> that you
>>> > didn't sketch how you delete messages from the queue. It's very
>>> important
>>> > to only delete those messages that are part of the successful
>>> checkpoint.
>>> > So you can't use PurgeQueue.
>>> >
>>> > *As mentioned earlier in the second point, we have already discussed
>>> the
>>> > mechanism. Please let us know if you have any concerns.*
>>> >
>>> > - I wonder if you need to use ReceiveRequestAttemptId. It looks like
>>> > it may be important for retries.
>>> >
>>> >
>>> >
>>> > *Since we use the AWS SDK library to establish connections and perform
>>> SQS
>>> > operations. This library already ensures retries for any transient
>>> issues,
>>> > with a default of three retries [3]. Additionally, AWS provides
>>> predefined
>>> > schemes to select from when creating the client.*
>>> > Please review and let us know your feedback on this.
>>> >
>>> >
>>> > *[1] *
>>> >
>>> https://nightlies.apache.org/flink/flink-docs-master/api/java/org/apache/flink/api/common/state/CheckpointListener.html
>>> > *[2] *
>>> >
>>> https://github.com/apache/flink/blob/master//flink-core/src/main/java/org/apache/flink/api/connector/source/DynamicParallelismInference.java#L29
>>> > *[3]*
>>> >
>>> https://github.com/aws/aws-sdk-java/blob/61d73631fac8535ad70666bbce9e70a1d2cea2ca/aws-java-sdk-core/src/main/java/com/amazonaws/retry/PredefinedRetryPolicies.java#L41
>>> >
>>> >
>>> >
>>> > Regards
>>> > Saurabh & Abhi
>>> >
>>> >
>>> >
>>> > On Tue, Sep 3, 2024 at 6:40 PM Arvid Heise <ar...@apache.org> wrote:
>>> >
>>> >> Sorry for being late to the party. I saw your call to vote and looked
>>> at
>>> >> the FLIP.
>>> >>
>>> >> First, most of the design is looking really good and it will be good
>>> to
>>> >> have another connector integrated into the AWS ecosystem. A couple of
>>> >> questions/remarks:
>>> >> 1) Since we only have 1 split, we should also limit the parallelism
>>> of the
>>> >> source accordingly. We could exploit the DynamicParallelismInference
>>> to
>>> >> effectively limit it to 1 unless the user explicitly overwrites.
>>> >> 2) If I haven't overlooked something obvious, then your exactly-once
>>> >> strategy will not work. There is unfortunately no guarantee
>>> >> that notifyCheckpointComplete is called at all before a failure
>>> happens.
>>> >> So
>>> >> you very likely get duplicate messages.
>>> >> Scanning the SQS documentation, I saw that you can read the
>>> SequenceNumber
>>> >> of the message. If you also store the latest number in the checkpoint
>>> >> state
>>> >> of the split, then you can discard all messages during recovery that
>>> are
>>> >> smaller. But I have never used SQS, so just take it as an inspiration.
>>> >> Also note that you didn't sketch how you delete messages from the
>>> queue.
>>> >> It's very important to only delete those messages that are part of the
>>> >> successful checkpoint. So you can't use PurgeQueue.
>>> >> 3) I wonder if you need to use ReceiveRequestAttemptId. It looks like
>>> it
>>> >> may be important for retries.
>>> >>
>>> >> Best,
>>> >>
>>> >> Arvid
>>> >>
>>> >> On Tue, Aug 20, 2024 at 11:24 AM Ahmed Hamdy <hamdy10...@gmail.com>
>>> >> wrote:
>>> >>
>>> >> > Hi Abhisagar and Saurabh
>>> >> > I have created the FLIP page and assigned it FLIP-477[1]. Feel free
>>> to
>>> >> > resume with the next steps.
>>> >> >
>>> >> > 1-
>>> >> >
>>> >> >
>>> >>
>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-+477+Amazon+SQS+Source+Connector
>>> >> > Best Regards
>>> >> > Ahmed Hamdy
>>> >> >
>>> >> >
>>> >> > On Tue, 20 Aug 2024 at 06:05, Abhisagar Khatri <
>>> >> > khatri.abhisaga...@gmail.com>
>>> >> > wrote:
>>> >> >
>>> >> > > Hi Flink Devs,
>>> >> > >
>>> >> > > Gentle Reminder for the request. We'd like to ask the
>>> PMC/Committers
>>> >> to
>>> >> > > transfer the content from the Amazon SQS Source Connector Google
>>> Doc
>>> >> [1]
>>> >> > > and assign a FLIP Number for us, which we can use further for
>>> voting.
>>> >> > > We are following the procedure outlined on the Flink Improvement
>>> >> Proposal
>>> >> > > Confluence page [2].
>>> >> > >
>>> >> > > [1]
>>> >> > >
>>> >> >
>>> >>
>>> https://docs.google.com/document/d/1lreo27jNh0LkRs1Mj9B3wj3itrzMa38D4_XGryOIFks/edit?usp=sharing
>>> >> > > [2]
>>> >> > >
>>> >> >
>>> >>
>>> https://cwiki.apache.org/confluence/display/FLINK/Flink+Improvement+Proposals#FlinkImprovementProposals-Process
>>> >> > >
>>> >> > > Regards,
>>> >> > > Abhi & Saurabh
>>> >> > >
>>> >> > >
>>> >> > > On Tue, Aug 13, 2024 at 12:50 PM Saurabh Singh <
>>> >> > saurabhsingh9...@gmail.com>
>>> >> > > wrote:
>>> >> > >
>>> >> > >> Hi Flink Devs,
>>> >> > >>
>>> >> > >> Thanks for all the feedback flink devs.
>>> >> > >>
>>> >> > >> Following the procedure outlined on the Flink Improvement
>>> Proposal
>>> >> > >> Confluence page [1], we kindly ask the PMC/Committers to
>>> transfer the
>>> >> > >> content from the Amazon SQS Source Connector Google Doc [2] and
>>> >> assign a
>>> >> > >> FLIP Number for us, which we will use for voting.
>>> >> > >>
>>> >> > >> [1]
>>> >> > >>
>>> >> >
>>> >>
>>> https://cwiki.apache.org/confluence/display/FLINK/Flink+Improvement+Proposals#FlinkImprovementProposals-Process
>>> >> > >> [2]
>>> >> > >>
>>> >> >
>>> >>
>>> https://docs.google.com/document/d/1lreo27jNh0LkRs1Mj9B3wj3itrzMa38D4_XGryOIFks/edit?usp=sharing
>>> >> > >>
>>> >> > >> Regards
>>> >> > >> Saurabh & Abhi
>>> >> > >>
>>> >> > >>
>>> >> > >> On Thu, Aug 8, 2024 at 5:12 PM Saurabh Singh <
>>> >> > saurabhsingh9...@gmail.com>
>>> >> > >> wrote:
>>> >> > >>
>>> >> > >>> Hi Ahmed,
>>> >> > >>>
>>> >> > >>> Yes, you're correct. Currently, we're utilizing the "record
>>> >> emitter" to
>>> >> > >>> send messages into the queue for deletion. However, for the
>>> actual
>>> >> > deletion
>>> >> > >>> process, which is dependent on the checkpoints, we've been
>>> using the
>>> >> > source
>>> >> > >>> reader class because it allows us to override the
>>> >> > notifyCheckpointComplete
>>> >> > >>> method.
>>> >> > >>>
>>> >> > >>> Regards
>>> >> > >>> Saurabh & Abhi
>>> >> > >>>
>>> >> > >>> On Wed, Aug 7, 2024 at 2:18 PM Ahmed Hamdy <
>>> hamdy10...@gmail.com>
>>> >> > wrote:
>>> >> > >>>
>>> >> > >>>> Hi Saurabh
>>> >> > >>>> Thanks for addressing, I see the FLIP is in much better state.
>>> >> > >>>> Could we specify where we queue messages for deletion, In my
>>> >> opinion
>>> >> > >>>> the record emitter is a good place for that where we delete
>>> >> messages
>>> >> > that
>>> >> > >>>> are forwarded to the next operator.
>>> >> > >>>> Other than that I don't have further comments.
>>> >> > >>>> Thanks again for the effort.
>>> >> > >>>>
>>> >> > >>>> Best Regards
>>> >> > >>>> Ahmed Hamdy
>>> >> > >>>>
>>> >> > >>>>
>>> >> > >>>> On Wed, 31 Jul 2024 at 10:34, Saurabh Singh <
>>> >> > saurabhsingh9...@gmail.com>
>>> >> > >>>> wrote:
>>> >> > >>>>
>>> >> > >>>>> Hi Ahmed,
>>> >> > >>>>>
>>> >> > >>>>> Thank you very much for the detailed, valuable review. Please
>>> find
>>> >> > our
>>> >> > >>>>> responses below:
>>> >> > >>>>>
>>> >> > >>>>>
>>> >> > >>>>> - In the FLIP you mention the split is going to be 1 sqs
>>> Queue,
>>> >> > >>>>> does this mean we would support reading from multiple
>>> queues?
>>> >> > This is also
>>> >> > >>>>> not clear in the implementation of `addSplitsBack` whether
>>> we
>>> >> are
>>> >> > planning
>>> >> > >>>>> to support multiple sqs topics or not.
>>> >> > >>>>>
>>> >> > >>>>> *Our current implementation assumes that each source reads
>>> from a
>>> >> > >>>>> single SQS queue. If you need to read from multiple SQS
>>> queues,
>>> >> you
>>> >> > can
>>> >> > >>>>> define multiple sources accordingly. We believe this approach
>>> is
>>> >> > clearer
>>> >> > >>>>> and more organized compared to having a single source switch
>>> >> between
>>> >> > >>>>> multiple queues. This design choice is based on weighing the
>>> >> > benefits, but
>>> >> > >>>>> we can support multiple queues per source if the need arises.*
>>> >> > >>>>>
>>> >> > >>>>> - Regarding Client creation, there has been some effort in
>>> the
>>> >> > >>>>> common `aws-util` module like createAwsSyncClient, we
>>> should
>>> >> > reuse that for
>>> >> > >>>>> `SqsClient` creation.
>>> >> > >>>>>
>>> >> > >>>>> *Thank you for bringing this to our attention. Yes, we will
>>> >> > >>>>> utilize the existing createClient methods available in the
>>> >> > libraries. Our
>>> >> > >>>>> goal is to avoid any code duplication on our end.*
>>> >> > >>>>>
>>> >> > >>>>>
>>> >> > >>>>> - On the same point for clients, Is there a reason the FLIP
>>> >> > >>>>> suggests async clients? sync clients have proven more
>>> stable
>>> >> and
>>> >> > the source
>>> >> > >>>>> threading model already guarantees no blocking by sync
>>> clients.
>>> >> > >>>>>
>>> >> > >>>>> *We were not aware of this, and we have been using async
>>> clients
>>> >> for
>>> >> > >>>>> our in-house use cases. However, since we already have sync
>>> >> clients
>>> >> > in the
>>> >> > >>>>> aws-util that ensure no blocking, we are in a good position.
>>> We
>>> >> will
>>> >> > use
>>> >> > >>>>> these sync clients during our development and testing
>>> efforts, and
>>> >> > we will
>>> >> > >>>>> share the results and keep the community updated.*
>>> >> > >>>>>
>>> >> > >>>>> - On mentioning threading, the FLIP doesn’t mention the
>>> fetcher
>>> >> > >>>>> manager. Is it going to be `SingleThreadFetcherManager`?
>>> Would
>>> >> it
>>> >> > be better
>>> >> > >>>>> to make the source reader extend the
>>> >> > SingleThreadedMultiplexReaderBase or
>>> >> > >>>>> are we going to implement a more simple version?
>>> >> > >>>>>
>>> >> > >>>>> *Yes, we are considering implementing
>>> >> > >>>>> SingleThreadMultiplexSourceReaderBase for the Reader. We have
>>> >> > included the
>>> >> > >>>>> implementation snippet in the FLIP for reference.*
>>> >> > >>>>>
>>> >> > >>>>> - The FLIP doesn’t mention schema deserialization or the
>>> >> > >>>>> recordEmitter implementation, Are we going to use
>>> >> > `deserializationSchema`
>>> >> > >>>>> or some sort of string to element converter? It is also not
>>> >> clear
>>> >> > form the
>>> >> > >>>>> builder example provided?
>>> >> > >>>>>
>>> >> > >>>>> *Yes, we plan to use the deserializationSchema and
>>> recordEmitter
>>> >> > >>>>> implementations. We have included sample code for these in the
>>> >> FLIP
>>> >> > for
>>> >> > >>>>> reference.*
>>> >> > >>>>>
>>> >> > >>>>> - Are the values mentioned in getSqsClientProperties
>>> >> recommended
>>> >> > >>>>> defaults? If so we should highlight that.
>>> >> > >>>>>
>>> >> > >>>>> *The defaults are not decided. These are just sample
>>> snapshots for
>>> >> > >>>>> example.*
>>> >> > >>>>>
>>> >> > >>>>> - Most importantly I am a bit skeptical regarding enforcing
>>> >> > >>>>> exactly-once semantics with side effects especially with
>>> >> > dependency on
>>> >> > >>>>> checkpointing configuration, could we add flags to disable
>>> and
>>> >> > disable by
>>> >> > >>>>> default if the checkpointing is not enabled?
>>> >> > >>>>>
>>> >> > >>>>> *During our initial design phase, we intended to enforce
>>> >> exactly-once
>>> >> > >>>>> semantics via checkpoints. However, you raise a valid point,
>>> and
>>> >> we
>>> >> > will
>>> >> > >>>>> make this a configurable feature for users. They can choose to
>>> >> > disable
>>> >> > >>>>> exactly-once semantics, accepting some duplicate processing
>>> >> > (at-least-once)
>>> >> > >>>>> as a trade-off. We have updated the FLIP to include support
>>> for
>>> >> this
>>> >> > >>>>> feature.*
>>> >> > >>>>>
>>> >> > >>>>> - I am not 100% convinced we should block FLIP itself on
>>> the
>>> >> > >>>>> FLIP-438 implementation but I echo the fact that there
>>> might be
>>> >> > some
>>> >> > >>>>> reusable code between the 2 submodules we should make use
>>> of.
>>> >> > >>>>>
>>> >> > >>>>> *Yes we echo the same. We fully understand the concern and
>>> are
>>> >> > >>>>> closely examining the SQS Sink implementation. We will
>>> ensure
>>> >> > there is no
>>> >> > >>>>> duplication of work or submodules. If any issues arise, we
>>> will
>>> >> > address
>>> >> > >>>>> them promptly.*
>>> >> > >>>>>
>>> >> > >>>>>
>>> >> > >>>>> Thank you for your valuable feedback on the FLIP. Your input
>>> is
>>> >> > >>>>> helping us refine and improve it significantly.
>>> >> > >>>>>
>>> >> > >>>>> [Main Proposal doc] -
>>> >> > >>>>>
>>> >> >
>>> >>
>>> https://docs.google.com/document/d/1lreo27jNh0LkRs1Mj9B3wj3itrzMa38D4_XGryOIFks/edit#heading=h.ci1rrcgbsvkl
>>> >> > >>>>>
>>> >> > >>>>> Regards
>>> >> > >>>>> Saurabh & Abhi
>>> >> > >>>>>
>>> >> > >>>>>
>>> >> > >>>>> On Sun, Jul 28, 2024 at 3:04 AM Ahmed Hamdy <
>>> hamdy10...@gmail.com
>>> >> >
>>> >> > >>>>> wrote:
>>> >> > >>>>>
>>> >> > >>>>>> Hi Saurabh
>>> >> > >>>>>> I think this is going to be a valuable addition which is
>>> needed.
>>> >> > >>>>>> I have a couple of comments
>>> >> > >>>>>> - In the FLIP you mention the split is going to be 1 sqs
>>> Queue,
>>> >> does
>>> >> > >>>>>> this
>>> >> > >>>>>> mean we would support reading from multiple queues? The
>>> builder
>>> >> > >>>>>> example
>>> >> > >>>>>> seems to support a single queue
>>> >> > >>>>>> SqsSource.<T>builder.setSqsUrl("
>>> >> > >>>>>> https://sqs.us-east-1.amazonaws.com/23145433/sqs-test";)
>>> >> > >>>>>> - This is also not clear in the implementation of
>>> `addSplitsBack`
>>> >> > >>>>>> whether
>>> >> > >>>>>> we are planning to support multiple sqs topics or not.
>>> >> > >>>>>> - Regarding Client creation, there has been some effort in
>>> the
>>> >> > common
>>> >> > >>>>>> `aws-util` module like createAwsSyncClient , we should reuse
>>> >> that
>>> >> > for
>>> >> > >>>>>> `SqsClient` creation.
>>> >> > >>>>>> - On the same point for clients, Is there a reason the FLIP
>>> >> suggests
>>> >> > >>>>>> async
>>> >> > >>>>>> clients? sync clients have proven more stable and the source
>>> >> > threading
>>> >> > >>>>>> model already guarantees no blocking by sync clients.
>>> >> > >>>>>> - On mentioning threading, the FLIP doesn't mention the
>>> fetcher
>>> >> > >>>>>> manager. Is
>>> >> > >>>>>> it going to be `SingleThreadFetcherManager`? Would it be
>>> better
>>> >> to
>>> >> > >>>>>> make the
>>> >> > >>>>>> source reader extend the SingleThreadedMultiplexReaderBase or
>>> >> are we
>>> >> > >>>>>> going
>>> >> > >>>>>> to implement a more simple version?
>>> >> > >>>>>> - The FLIP doesn't mention schema deserialization or the
>>> >> > recordEmitter
>>> >> > >>>>>> implementation, Are we going to use `deserializationSchema`
>>> or
>>> >> some
>>> >> > >>>>>> sort of
>>> >> > >>>>>> string to element converter? It is also not clear form the
>>> >> builder
>>> >> > >>>>>> example
>>> >> > >>>>>> provided?
>>> >> > >>>>>> - Are the values mentioned in getSqsClientProperties
>>> recommended
>>> >> > >>>>>> defaults?
>>> >> > >>>>>> If so we should highlight that
>>> >> > >>>>>> - Most importantly I am a bit skeptical regarding enforcing
>>> >> > >>>>>> exactly-once
>>> >> > >>>>>> semantics with side effects especially with dependency on
>>> >> > >>>>>> checkpointing
>>> >> > >>>>>> configuration, could we add flags to disable and disable by
>>> >> default
>>> >> > >>>>>> if the
>>> >> > >>>>>> checkpointing is not enabled?
>>> >> > >>>>>>
>>> >> > >>>>>> I am not 100% convinced we should block FLIP itself on the
>>> >> FLIP-438
>>> >> > >>>>>> implementation but I echo the fact that there might be some
>>> >> reusable
>>> >> > >>>>>> code
>>> >> > >>>>>> between the 2 submodules we should make use of.
>>> >> > >>>>>>
>>> >> > >>>>>> Best Regards
>>> >> > >>>>>> Ahmed Hamdy
>>> >> > >>>>>>
>>> >> > >>>>>>
>>> >> > >>>>>> On Fri, 26 Jul 2024 at 17:55, Saurabh Singh <
>>> >> > >>>>>> saurabhsingh9...@gmail.com>
>>> >> > >>>>>> wrote:
>>> >> > >>>>>>
>>> >> > >>>>>> > Hi Li Wang,
>>> >> > >>>>>> >
>>> >> > >>>>>> > Thanks for the review and appreciate your feedback.
>>> >> > >>>>>> > I completely understand your concern and agree with it. Our
>>> >> goal
>>> >> > is
>>> >> > >>>>>> to
>>> >> > >>>>>> > provide users of connectors with a consistent and coherent
>>> >> > >>>>>> ecosystem, free
>>> >> > >>>>>> > of any issues.
>>> >> > >>>>>> > To ensure that, we are closely monitoring/reviewing the SQS
>>> >> Sink
>>> >> > >>>>>> > implementation work by Dhingra. Our development work will
>>> >> commence
>>> >> > >>>>>> once the
>>> >> > >>>>>> > AWS Sink is near completion or completed. This approach
>>> ensures
>>> >> > >>>>>> that we
>>> >> > >>>>>> > take in the new learnings, do not duplicate any core
>>> modules
>>> >> and
>>> >> > >>>>>> allow us
>>> >> > >>>>>> > to save valuable time.
>>> >> > >>>>>> >
>>> >> > >>>>>> > In the meantime, we would like to keep the discussion and
>>> >> process
>>> >> > >>>>>> active on
>>> >> > >>>>>> > this FLIP. Gaining valuable community feedback (which is
>>> >> helping
>>> >> > >>>>>> us) will
>>> >> > >>>>>> > help us address any potential gaps in the source connector
>>> >> design
>>> >> > >>>>>> and
>>> >> > >>>>>> > finalize it. Behind the scenes, we are already designing
>>> and
>>> >> > >>>>>> pre-planning
>>> >> > >>>>>> > our development work to adhere to feedback/best practices/
>>> >> faster
>>> >> > >>>>>> delivery
>>> >> > >>>>>> > when we implement this FLIP.
>>> >> > >>>>>> > Please share your thoughts on this.
>>> >> > >>>>>> >
>>> >> > >>>>>> > Regards
>>> >> > >>>>>> > Saurabh
>>> >> > >>>>>> >
>>> >> > >>>>>> > On Fri, Jul 26, 2024 at 5:52 PM Li Wang <
>>> >> liwang505...@gmail.com>
>>> >> > >>>>>> wrote:
>>> >> > >>>>>> >
>>> >> > >>>>>> > > Hi Saurabh,
>>> >> > >>>>>> > >
>>> >> > >>>>>> > > Thanks for the FLIP. Given the ongoing effort to
>>> implement
>>> >> the
>>> >> > >>>>>> SQS sink
>>> >> > >>>>>> > > connector (
>>> >> > >>>>>> > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://cwiki.apache.org/confluence/display/FLINK/FLIP-438%3A+Amazon+SQS+Sink+Connector
>>> >> > >>>>>> > > ),
>>> >> > >>>>>> > > it is important to consider how the SQS source connector
>>> >> > supports
>>> >> > >>>>>> this
>>> >> > >>>>>> > > development, ensuring a unified framework for AWS
>>> integration
>>> >> > >>>>>> that avoids
>>> >> > >>>>>> > > capacity overlap and maximizes synchronization. Since the
>>> >> > >>>>>> implementation
>>> >> > >>>>>> > by
>>> >> > >>>>>> > > Dhingra is still WIP, I would suggest taking that into
>>> >> account
>>> >> > >>>>>> and keep
>>> >> > >>>>>> > > this FLIP as an extension of that FLIP which could be
>>> taken
>>> >> up
>>> >> > >>>>>> once the
>>> >> > >>>>>> > SQS
>>> >> > >>>>>> > > Sink FLIP is fully implemented. If not, this may lead to
>>> >> > >>>>>> doublework in
>>> >> > >>>>>> > > multiple identity-related modules and structure of the
>>> >> overall
>>> >> > SQS
>>> >> > >>>>>> > > connector codebase. What do you think?
>>> >> > >>>>>> > >
>>> >> > >>>>>> > > Thanks
>>> >> > >>>>>> > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> > > On Thursday, July 25, 2024, Saurabh Singh <
>>> >> > >>>>>> saurabhsingh9...@gmail.com>
>>> >> > >>>>>> > > wrote:
>>> >> > >>>>>> > >
>>> >> > >>>>>> > > > Hi Samrat,
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > Thanks for the review and feedback.
>>> >> > >>>>>> > > > We have evaluated all the three points. Please find the
>>> >> > answers
>>> >> > >>>>>> below:
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > 1. AWS has announced JSON protocol support in SQS [1].
>>> Can
>>> >> you
>>> >> > >>>>>> shed
>>> >> > >>>>>> > some
>>> >> > >>>>>> > > > light on how different protocols will be supported?
>>> >> > >>>>>> > > > - We will utilize the AWS client library to connect
>>> with
>>> >> the
>>> >> > >>>>>> AWS SQS
>>> >> > >>>>>> > > > Service. Versions beyond 2.21.19 now support JSON, so
>>> >> simply
>>> >> > >>>>>> upgrading
>>> >> > >>>>>> > > the
>>> >> > >>>>>> > > > client library will suffice for the protocol switch.
>>> >> However,
>>> >> > >>>>>> from the
>>> >> > >>>>>> > > > connector's perspective, we do not anticipate any
>>> changes
>>> >> in
>>> >> > our
>>> >> > >>>>>> > > > communication process, as it is handled by the client
>>> >> library.
>>> >> > >>>>>> [4]
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > 2. AWS SQS has two types of queues [2]. What are the
>>> >> > >>>>>> implementation
>>> >> > >>>>>> > > detail
>>> >> > >>>>>> > > > differences for the source connector?
>>> >> > >>>>>> > > > - SQS Connector is indifferent to the customer's
>>> choice of
>>> >> > >>>>>> Queue type.
>>> >> > >>>>>> > If
>>> >> > >>>>>> > > > out-of-order messages are a concern, it will be the
>>> >> > >>>>>> responsibility of
>>> >> > >>>>>> > the
>>> >> > >>>>>> > > > application code or main job logic to manage this.
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > 3. Will the SQS source connector implement any kind of
>>> >> > >>>>>> callbacks [3] on
>>> >> > >>>>>> > > > success to offer any kind of guarantee?
>>> >> > >>>>>> > > > - We have proposed deleting SQS messages using the
>>> >> > notification
>>> >> > >>>>>> > provided
>>> >> > >>>>>> > > by
>>> >> > >>>>>> > > > the checkpoint framework on checkpoint completion. Thus
>>> >> > >>>>>> providing
>>> >> > >>>>>> > exactly
>>> >> > >>>>>> > > > once guarantee.[5] Additionally, when deleting
>>> messages, we
>>> >> > will
>>> >> > >>>>>> > monitor
>>> >> > >>>>>> > > > the API call responses and log any failures, along with
>>> >> > >>>>>> providing
>>> >> > >>>>>> > > > observability through appropriate metrics.
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > [1]
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://aws.amazon.com/about-aws/whats-new/2023/11/amazon-sqs-support-json-protocol/
>>> >> > >>>>>> > > > [2]
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-queue-types.html
>>> >> > >>>>>> > > > [3]
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://docs.aws.amazon.com/step-functions/latest/dg/callback-task-sample-sqs.html
>>> >> > >>>>>> > > > [4]
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-json-faqs.html#json-protocol-getting-started
>>> >> > >>>>>> > > > [5]
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://docs.google.com/document/d/1lreo27jNh0LkRs1Mj9B3wj3itrzMa38D4_XGryOIFks/edit#heading=h.jdcikzojx5d9
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > [Main Proposal doc] -
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://docs.google.com/document/d/1lreo27jNh0LkRs1Mj9B3wj3itrzMa38D4_XGryOIFks/edit#heading=h.ci1rrcgbsvkl
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > Please feel free to reach out if you have more
>>> feedback.
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > Regards
>>> >> > >>>>>> > > > Saurabh & Abhi
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > On Wed, Jul 24, 2024 at 8:52 AM Samrat Deb <
>>> >> > >>>>>> decordea...@gmail.com>
>>> >> > >>>>>> > > wrote:
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > > > > Hi Saurabh,
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > > Thank you for sharing the FLIP for the SQS source
>>> >> connector.
>>> >> > >>>>>> An SQS
>>> >> > >>>>>> > > > source
>>> >> > >>>>>> > > > > connector will be a great addition to the Flink
>>> >> ecosystem,
>>> >> > as
>>> >> > >>>>>> there
>>> >> > >>>>>> > is
>>> >> > >>>>>> > > a
>>> >> > >>>>>> > > > > growing demand for SQS source/sink integration.
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > > I have a few queries:
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > > 1. AWS has announced JSON protocol support in SQS
>>> [1].
>>> >> Can
>>> >> > >>>>>> you shed
>>> >> > >>>>>> > > some
>>> >> > >>>>>> > > > > light on how different protocols will be supported?
>>> >> > >>>>>> > > > > 2. AWS SQS has two types of queues [2]. What are the
>>> >> > >>>>>> implementation
>>> >> > >>>>>> > > > detail
>>> >> > >>>>>> > > > > differences for the source connector?
>>> >> > >>>>>> > > > > 3. Will the SQS source connector implement any kind
>>> of
>>> >> > >>>>>> callbacks [3]
>>> >> > >>>>>> > on
>>> >> > >>>>>> > > > > success to offer any kind of guarantee?
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > > [1]
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://aws.amazon.com/about-aws/whats-new/2023/11/amazon-sqs-support-json-protocol/
>>> >> > >>>>>> > > > > [2]
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://docs.aws.amazon.com/AWSSimpleQueueService/latest/SQSDeveloperGuide/sqs-queue-types.html
>>> >> > >>>>>> > > > > [3]
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://docs.aws.amazon.com/step-functions/latest/dg/callback-task-sample-sqs.html
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > > Bests,
>>> >> > >>>>>> > > > > Samrat
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > > On Fri, 19 Jul 2024 at 9:53 PM, Saurabh Singh <
>>> >> > >>>>>> > > > saurabhsingh9...@gmail.com>
>>> >> > >>>>>> > > > > wrote:
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > > > > Hi Fink Devs,
>>> >> > >>>>>> > > > > >
>>> >> > >>>>>> > > > > > Our team has been working on migrating various data
>>> >> > >>>>>> pipelines to
>>> >> > >>>>>> > > Flink
>>> >> > >>>>>> > > > to
>>> >> > >>>>>> > > > > > leverage the benefits of exactly-once processing,
>>> >> > >>>>>> checkpointing,
>>> >> > >>>>>> > and
>>> >> > >>>>>> > > > > > stateful computing. We have several use cases built
>>> >> around
>>> >> > >>>>>> the AWS
>>> >> > >>>>>> > > SQS
>>> >> > >>>>>> > > > > > Service. For this migration, we have developed an
>>> SQS
>>> >> > Source
>>> >> > >>>>>> > > Connector,
>>> >> > >>>>>> > > > > > which enables us to run both stateless and stateful
>>> >> > >>>>>> Flink-based
>>> >> > >>>>>> > jobs.
>>> >> > >>>>>> > > > > >
>>> >> > >>>>>> > > > > > We believe that this SQS Source Connector would be
>>> a
>>> >> > >>>>>> valuable
>>> >> > >>>>>> > > addition
>>> >> > >>>>>> > > > to
>>> >> > >>>>>> > > > > > the existing connector set. Therefore, we propose a
>>> >> FLIP
>>> >> > to
>>> >> > >>>>>> include
>>> >> > >>>>>> > > it.
>>> >> > >>>>>> > > > > >
>>> >> > >>>>>> > > > > > For more information, please refer to the FLIP
>>> >> document.
>>> >> > >>>>>> > > > > >
>>> >> > >>>>>> > > > > >
>>> >> > >>>>>> > > > > >
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> >
>>> >>
>>> https://docs.google.com/document/d/1lreo27jNh0LkRs1Mj9B3wj3itrzMa38D4_XGryOIFks/edit?usp=sharing
>>> >> > >>>>>> > > > > >
>>> >> > >>>>>> > > > > > Thanks
>>> >> > >>>>>> > > > > > Saurabh & Abhi
>>> >> > >>>>>> > > > > >
>>> >> > >>>>>> > > > >
>>> >> > >>>>>> > > >
>>> >> > >>>>>> > >
>>> >> > >>>>>> >
>>> >> > >>>>>>
>>> >> > >>>>>
>>> >> >
>>> >>
>>> >
>>>
>>
