fix: no timezone while parsing (#2671)

rudderlabs · Nov 28, 2022 · 9f006e5 · 9f006e5
1 parent 0941cc0
commit 9f006e5
Show file tree

Hide file tree

Showing 3 changed files with 17 additions and 4 deletions.
diff --git a/enterprise/replay/dumpsloader.go b/enterprise/replay/dumpsloader.go
@@ -54,8 +54,12 @@ func getMinMaxCreatedAt(key string) (int64, int64, error) {
 		return minJobCreatedAt, maxJobCreatedAt, fmt.Errorf("%s 's parse with _ gave tokens more than 3. Expected 3", key)
 	}
 	keyTokens = strings.Split(keyTokens[2], ".")
-	if len(keyTokens) != 6 {
-		return minJobCreatedAt, maxJobCreatedAt, fmt.Errorf("%s 's parse with . gave tokens more than 6. Expected 6", keyTokens[2])
+	if len(keyTokens) > 7 {
+		return minJobCreatedAt, maxJobCreatedAt, fmt.Errorf("%s 's parse with . gave tokens more than 7. Expected 6 or 7", keyTokens[2])
+	}
+
+	if len(keyTokens) < 6 { // for backward compatibility TODO: remove this check after some time
+		return minJobCreatedAt, maxJobCreatedAt, fmt.Errorf("%s 's parse with . gave tokens less than 6. Expected 6 or 7", keyTokens[2])
 	}
 	minJobCreatedAt, err = strconv.ParseInt(keyTokens[3], 10, 64)
 	if err != nil {

diff --git a/enterprise/replay/sourceWorker.go b/enterprise/replay/sourceWorker.go
@@ -120,7 +120,8 @@ func (worker *SourceWorkerT) replayJobsInFile(ctx context.Context, filePath stri
 		copy(copyLineBytes, lineBytes)
 
 		if transformationVersionID == "" {
-			createdAt, err := time.Parse(misc.POSTGRESTIMEFORMATPARSE, gjson.GetBytes(copyLineBytes, worker.getFieldIdentifier(createdAt)).String())
+			timeStamp := gjson.GetBytes(copyLineBytes, worker.getFieldIdentifier(createdAt)).String()
+			createdAt, err := time.Parse(misc.NOTIMEZONEFORMATPARSE, getFormattedTimeStamp(timeStamp))
 			if err != nil {
 				worker.log.Errorf("failed to parse created at: %s", err)
 				continue
@@ -178,7 +179,7 @@ func (worker *SourceWorkerT) replayJobsInFile(ctx context.Context, filePath stri
 				worker.log.Errorf("Error getting created at from transformer output: %v", err)
 				continue
 			}
-			createdAt, err := time.Parse(misc.POSTGRESTIMEFORMATPARSE, createdAtString)
+			createdAt, err := time.Parse(misc.NOTIMEZONEFORMATPARSE, getFormattedTimeStamp(createdAtString))
 			if err != nil {
 				worker.log.Errorf("failed to parse created at: %s", err)
 				continue
@@ -197,6 +198,7 @@ func (worker *SourceWorkerT) replayJobsInFile(ctx context.Context, filePath stri
 				Parameters:   params,
 				CustomVal:    ev.Output[worker.getFieldIdentifier(customVal)].(string),
 				EventPayload: destEventJSON,
+				WorkspaceId:  ev.Output[worker.getFieldIdentifier(workspaceID)].(string),
 			}
 			jobs = append(jobs, &job)
 		}
@@ -256,9 +258,15 @@ func (worker *SourceWorkerT) getFieldIdentifier(field string) string {
 		return "CustomVal"
 	case eventPayload:
 		return "EventPayload"
+	case workspaceID:
+		return "WorkspaceID"
 	case createdAt:
 		return "CreatedAt"
 	default:
 		return ""
 	}
 }
+
+func getFormattedTimeStamp(timeStamp string) string {
+	return strings.Split(strings.TrimSuffix(timeStamp, "Z"), ".")[0]
+}
diff --git a/utils/misc/misc.go b/utils/misc/misc.go
@@ -61,6 +61,7 @@ const (
 	// RFC3339Milli with milli sec precision
 	RFC3339Milli            = "2006-01-02T15:04:05.000Z07:00"
 	POSTGRESTIMEFORMATPARSE = "2006-01-02T15:04:05Z"
+	NOTIMEZONEFORMATPARSE   = "2006-01-02T15:04:05"
 )
 
 const (