From 2fb4ed0dd05afec408432d271f731d66c586191a Mon Sep 17 00:00:00 2001
From: juliannzhou <65748924+juliannzhou@users.noreply.github.com>
Date: Wed, 5 Jun 2024 06:54:24 -0400
Subject: [PATCH] [DSMS-19] Fix SNS topic format issue (#2725)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Co-authored-by: Dario Castañé <dario.castane@datadoghq.com>
---
 internal/datastreams/pathway.go | 14 ++++++++++++++
 1 file changed, 14 insertions(+)

diff --git a/internal/datastreams/pathway.go b/internal/datastreams/pathway.go
index 5ff05de272..e986ecb990 100644
--- a/internal/datastreams/pathway.go
+++ b/internal/datastreams/pathway.go
@@ -17,6 +17,17 @@ import (
 
 var hashableEdgeTags = map[string]struct{}{"event_type": {}, "exchange": {}, "group": {}, "topic": {}, "type": {}, "direction": {}}
 
+func isValidArn(arn string) bool {
+	separators := strings.Count(arn, ":")
+	if separators < 5 {
+		return false
+	}
+	if strings.HasPrefix(arn, "arn:") {
+		return true
+	}
+	return false
+}
+
 func isWellFormedEdgeTag(t string) bool {
 	if i := strings.IndexByte(t, ':'); i != -1 {
 		if j := strings.LastIndexByte(t, ':'); j == i {
@@ -24,6 +35,9 @@ func isWellFormedEdgeTag(t string) bool {
 				return true
 			}
 		}
+		if t[:i] == "topic" && isValidArn(t) {
+			return true
+		}
 	}
 	return false
 }