aws-samples / aws-samples/spark-streaming-sql-s3-connector
[Feature request] Add capability to read S3 notifications sent to SQS via. SNS
- Dominant language
- Scala
- Stars
- 16
- Forks
- 4
- PR merge metrics
- No merged PRs in 30d
Description
In order to support fanout, many systems have SNS notifications tied to S3 file arrival notifications, which then chain messages to SQS queues for consumption by systems like Spark. It would be useful for the framework to detect when this is the case, and parse the notification from within the SNS message body.
Simple fix in [AsyncSQSClientImpl.java](https://github.com/aws-samples/spark-streaming-sql-s3-connector/blob/fbc6de2cee2017e2a72b22e13bcf699be6298bba/src/main/scala/com/amazonaws/spark/sql/streaming/connector/client/AsyncSqsClientImpl.scala#L321-L360)
```
val messageJson = parse(messageBody).extract[JValue]
```
Becomes
```
var messageJson = parse(messageBody).extract[JValue]
if(messageJson \ "Message" != Nil) {
messageJson = parse( (messageJson \ "Message").extract[String] ).extract[JValue]
}
```
Sample S3 notification delivered to SQS via SNS (SNS notification in the "Message" field):
```
{
"Type" : "Notification",
"MessageId" : "0216b511-4ade-5a33-8ebb-ea1b34659891",
"TopicArn" : "arn:aws:sns:us-east-1::",
"Subject" : "Amazon S3 Notification",
"Message" : "{\"Records\":[{\"eventVersion\":\"2.1\",\"eventSource\":\"aws:s3\",\"awsRegion\":\"us-east-1\",\"eventTime\":\"2024-07-12T00:44:42.504Z\",\"eventName\":\"ObjectCreated:Put\",\"userIdentity\":{\"principalId\":\"\"},\"requestParameters\":},\"object\":{\"key\":\"/.json.gz.parquet.gz\",\"}}]}",
"Timestamp" : "2024-07-12T00:44:43.191Z",
"SignatureVersion" : "1",
"Signature" : "",
"SigningCertURL" : "",
"UnsubscribeURL" : ""
}
```
I am curious of the maintainer's thoughts, and am will to send a PR to address.
Contributor guide
Assessment
This issue has not been assessed yet.