Hi there, I'm following up on FLIP-477 (Amazon SQS Source Connector) [1].
I can see that the design was discussed and later voted and approved in this thread [2], but it looks like there hasn't been any activity on the thread or the umbrella JIRA (FLINK-36432) [3] for close to two years now. I think the source connector work was intentionally deferred until the SQS Sink connector (FLIP-438) [4] completes, to avoid duplicating effort on shared code. The sink has since shipped [5], and there are plenty of use cases that consume messages from SQS in addition to producing to it, so the source connector would fill the gap. I wanted to check if anyone is still actively working on the implementation? If not, I'd be happy to volunteer to pick this up and drive it forward, following the design as it was approved in the vote. Thanks, Wenhao [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-477+Amazon+SQS+Source+Connector [2] https://www.mail-archive.com/[email protected]/msg78319.html [3] https://issues.apache.org/jira/browse/FLINK-36432 [4] https://cwiki.apache.org/confluence/display/FLINK/FLIP-438:+Amazon+SQS+Sink+Connector [5] https://github.com/apache/flink-connector-aws/pull/141 On 2024/09/27 07:13:04 Arvid Heise wrote: > I have updated the FLIP [1] (please double-check) and will add my +1 now to > the voting. > > Best, > > Arvid > > [1] > https://cwiki.apache.org/confluence/display/FLINK/FLIP-477+Amazon+SQS+Source+Connector > > On Wed, Sep 25, 2024 at 2:21 PM Saurabh Singh <[email protected]> > wrote: > > > 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/lates [message truncated...]
