Merge branch 'main' into fix/bq-soft-delete-updates-in-batch

PeerDB-io · Dec 21, 2023 · a0f8f74 · a0f8f74
2 parents eac5b85 + e1b59cf
commit a0f8f74
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 2 deletions.
diff --git a/flow/connectors/utils/avro/avro_writer.go b/flow/connectors/utils/avro/avro_writer.go
@@ -216,8 +216,9 @@ func (p *peerDBOCFWriter) WriteRecordsToS3(bucketName, key string, s3Creds utils
 		Body:   r,
 	})
 	if err != nil {
-		slog.Error("failed to upload file: ", slog.Any("error", err))
-		return nil, fmt.Errorf("failed to upload file: %w", err)
+		s3Path := "s3://" + bucketName + "/" + key
+		slog.Error("failed to upload file: ", slog.Any("error", err), slog.Any("s3_path", s3Path))
+		return nil, fmt.Errorf("failed to upload file to path %s: %w", s3Path, err)
 	}
 
 	slog.Info("file uploaded to" + result.Location)

diff --git a/flow/e2e/s3/cdc_s3_test.go b/flow/e2e/s3/cdc_s3_test.go
@@ -22,6 +22,11 @@ func (s *PeerFlowE2ETestSuiteS3) Test_Complete_Simple_Flow_S3() {
 	env := s.NewTestWorkflowEnvironment()
 	e2e.RegisterWorkflowsAndActivities(env, s.T())
 
+	setupErr := s.setupS3("s3")
+	if setupErr != nil {
+		s.Fail("failed to setup S3", setupErr)
+	}
+
 	srcTableName := s.attachSchemaSuffix("test_simple_flow_s3")
 	dstTableName := fmt.Sprintf("%s.%s", "peerdb_test_s3", "test_simple_flow_s3")
 	flowJobName := s.attachSuffix("test_simple_flow_s3")