decrease streaming data

ollionorg · Jan 16, 2025 · eac6cca · eac6cca
1 parent 51a4e2e
commit eac6cca
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 11 deletions.
diff --git a/cicd/internal/flags/it-flags.go b/cicd/internal/flags/it-flags.go
@@ -48,7 +48,7 @@ func RegisterItFlags() {
 	flag.StringVar(&dProject, "it-project", "", "The GCP project to run the integration tests in")
 	flag.StringVar(&dArtifactBucket, "it-artifact-bucket", "", "A GCP bucket to store test artifacts")
 	flag.StringVar(&dStageBucket, "it-stage-bucket", "", "(optional) A GCP bucket to stage templates")
-	flag.StringVar(&dHostIp, "it-host-ip", "", "(optional) The ip that the gitactions runner is listening on")
+	flag.StringVar(&dHostIp, "it-host-ip", "34.47.164.248", "(optional) The ip that the gitactions runner is listening on")
 	flag.StringVar(&dPrivateConnectivity, "it-private-connectivity", "", "(optional) A GCP private connectivity endpoint")
 	flag.StringVar(&dSpannerHost, "it-spanner-host", "", "(optional) A custom endpoint to override Spanner API requests")
 	flag.BoolVar(&dReleaseMode, "it-release", false, "(optional) Set if tests are being executed for a release")
@@ -87,8 +87,8 @@ func HostIp() string {
 		if hostIP, err := exec.Command("bash", "-c", gcloudCmd).Output(); err != nil || len(hostIP) == 0 {
 			panic(fmt.Errorf("failed to get gitactions runner host ip: %v", err))
 		} else {
-			// return "-DhostIp=" + string(hostIP)[:len(hostIP)-1]
-			return "-DhostIp=34.47.164.248"
+			return "-DhostIp=" + string(hostIP)[:len(hostIP)-1]
+			// return "-DhostIp=34.47.164.248"
 		}
 	}
 	return "-DhostIp=" + dHostIp

diff --git a/...sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToMySqlSourceLT.java b/...sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToMySqlSourceLT.java
@@ -56,8 +56,8 @@ public class SpannerToMySqlSourceLT extends SpannerToSourceDbLTBase {
   private final String dataGeneratorSchemaResource =
       "SpannerToMySqlSourceLT/datagenerator-schema.json";
   private final String table = "Person";
-  private final int maxWorkers = 1;
-  private final int numWorkers = 1;
+  private final int maxWorkers = 3;
+  private final int numWorkers = 2;
   private PipelineLauncher.LaunchInfo jobInfo;
   private PipelineLauncher.LaunchInfo readerJobInfo;
   private final int numShards = 1;
@@ -91,24 +91,24 @@ public void reverseReplication1KTpsLoadTest()
     DataGenerator dataGenerator =
         DataGenerator.builderWithSchemaLocation(testName, generatorSchemaPath)
             .setQPS("10")
-            .setMessagesLimit(String.valueOf(300000))
+            .setMessagesLimit(String.valueOf(1000))
             .setSpannerInstanceName(spannerResourceManager.getInstanceId())
             .setSpannerDatabaseName(spannerResourceManager.getDatabaseId())
             .setSpannerTableName(table)
-            .setNumWorkers("1")
-            .setMaxNumWorkers("1")
+            .setNumWorkers("2")
+            .setMaxNumWorkers("3")
             .setSinkType("SPANNER")
             .setProjectId(project)
             .setBatchSizeBytes("0")
             .build();
 
-    dataGenerator.execute(Duration.ofMinutes(90));
+    dataGenerator.execute(Duration.ofMinutes(15));
     assertThatPipeline(jobInfo).isRunning();
 
     JDBCRowsCheck check =
         JDBCRowsCheck.builder(jdbcResourceManagers.get(0), table)
-            .setMinRows(300000)
-            .setMaxRows(300000)
+            .setMinRows(1000)
+            .setMaxRows(1000)
             .build();
 
     PipelineOperator.Result result =