abhishekd0907 commented on a change in pull request #97:
URL: https://github.com/apache/bahir/pull/97#discussion_r463742385
##########
File path:
sql-streaming-sqs/src/main/scala/org/apache/spark/sql/streaming/sqs/SqsClient.scala
##########
@@ -131,13 +131,24 @@ class SqsClient(sourceOptions: SqsSourceOptions,
}
}
+ private def extractS3Message(parsedBody: JValue): JValue = {
+ implicit val formats = DefaultFormats
+ sourceOptions.messageWrapper match {
+ case sourceOptions.S3MessageWrapper.None => parsedBody
+ case sourceOptions.S3MessageWrapper.SNS => parse((parsedBody \
"Message").extract[String])
Review comment:
I see. There's a retry on failure mechanism for error in fetching
messages from SQS. Maybe we should have an option for retry on failure in case
of an error in parsing messages too. Of course, it's not related to this and
can be done as a part of a separate PR.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
