IT fixes

Author: Deep1998

v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/migrations/utils/SchemaUtils.java

@@ -122,9 +122,9 @@ public static SourceSchema buildSourceSchemaFromSessionFile(String sessionFile)
               Map<String, Object> colMap = (Map<String, Object>) colDefs.get(colId);
               Map<String, Object> typeMap = (Map<String, Object>) colMap.get("Type");
               String typeName = (String) typeMap.get("Name");
-              Integer size = null;
+              Long size = null;
               if (typeMap.get("Len") instanceof Number) {
-                size = ((Number) typeMap.get("Len")).intValue();
+                size = ((Number) typeMap.get("Len")).longValue();
               }
               SourceColumn.Builder colBuilder =
                   SourceColumn.builder(dbType)

v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/sourceddl/MySqlInformationSchemaScanner.java

@@ -69,7 +69,7 @@ protected List<SourceColumn> scanColumns(String tableName, String schema) throws
         // Handle size/precision/scale
         String maxLength = rs.getString("character_maximum_length");
         if (maxLength != null) {
-          columnBuilder.size(Integer.parseInt(maxLength));
+          columnBuilder.size(Long.parseLong(maxLength));
         }
 
         String precision = rs.getString("numeric_precision");

v2/spanner-common/src/main/java/com/google/cloud/teleport/v2/spanner/sourceddl/SourceColumn.java

@@ -35,7 +35,7 @@ public abstract class SourceColumn implements Serializable {
   public abstract boolean isPrimaryKey();
 
   @Nullable
-  public abstract Integer size();
+  public abstract Long size();
 
   @Nullable
   public abstract Integer precision();
@@ -68,7 +68,7 @@ public abstract static class Builder {
 
     public abstract Builder isPrimaryKey(boolean isPrimaryKey);
 
-    public abstract Builder size(Integer size);
+    public abstract Builder size(Long size);
 
     public abstract Builder precision(Integer precision);
 

v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/sourceddl/MySqlInformationSchemaScannerTest.java

@@ -91,7 +91,7 @@ public void testScanSingleTable() throws SQLException {
     assertEquals("INT", column.type());
     assertEquals(false, column.isNullable());
     assertEquals(true, column.isPrimaryKey());
-    assertEquals(Integer.valueOf(10), column.size());
+    assertEquals(Long.valueOf(10L), column.size());
     assertEquals(1, table.primaryKeyColumns().size());
     assertEquals("id", table.primaryKeyColumns().get(0));
   }

v2/spanner-common/src/test/java/com/google/cloud/teleport/v2/spanner/sourceddl/SourceColumnTest.java

@@ -31,7 +31,7 @@ public void testBuilderAndProperties() {
             .type("INT")
             .isNullable(false)
             .isPrimaryKey(true)
-            .size(10)
+            .size(10L)
             .precision(5)
             .scale(0)
             .columnOptions(ImmutableList.of("AUTO_INCREMENT"))
@@ -40,7 +40,7 @@ public void testBuilderAndProperties() {
     assertEquals("INT", column.type());
     assertEquals(false, column.isNullable());
     assertEquals(true, column.isPrimaryKey());
-    assertEquals(Integer.valueOf(10), column.size());
+    assertEquals(Long.valueOf(10L), column.size());
     assertEquals(Integer.valueOf(5), column.precision());
     assertEquals(Integer.valueOf(0), column.scale());
     assertEquals(ImmutableList.of("AUTO_INCREMENT"), column.columnOptions());

v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/SpannerToSourceDb.java

@@ -600,6 +600,8 @@ public static PipelineResult run(Options options) {
     }
     SourceInformationSchemaScanner scanner = null;
     try {
+      SourceProcessorFactory.initializeConnectionHelper(
+          options.getSourceType(), shards, connectionPoolSizePerWorker);
       if (options.getSourceType().equals(MYSQL_SOURCE_TYPE)) {
         Connection connection =
             (Connection)

v2/spanner-to-sourcedb/src/main/java/com/google/cloud/teleport/v2/templates/dbutils/processor/SourceProcessorFactory.java

@@ -140,7 +140,7 @@ private static IConnectionHelper getConnectionHelper(String source)
                     "Invalid source type for connection helper: " + source));
   }
 
-  private static void initializeConnectionHelper(
+  public static void initializeConnectionHelper(
       String source, List<Shard> shards, int maxConnections) throws UnsupportedSourceException {
     IConnectionHelper connectionHelper = getConnectionHelper(source);
     if (!connectionHelper.isConnectionPoolInitialized()) {

v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToCassandraSourceDBCustomTransformationIT.java

@@ -114,7 +114,8 @@ public void setUp() throws IOException, InterruptedException {
                 null,
                 null,
                 customTransformation,
-                CASSANDRA_SOURCE_TYPE);
+                CASSANDRA_SOURCE_TYPE,
+                null);
       }
     }
   }

v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToCassandraSourceDbIT.java

@@ -142,7 +142,8 @@ public void setUp() throws IOException {
                 null,
                 null,
                 null,
-                CASSANDRA_SOURCE_TYPE);
+                CASSANDRA_SOURCE_TYPE,
+                null);
       }
     }
   }

v2/spanner-to-sourcedb/src/test/java/com/google/cloud/teleport/v2/templates/SpannerToCassandraSourceDbMaxColumnsIT.java

@@ -33,6 +33,7 @@
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
+import java.util.Map;
 import org.apache.beam.it.cassandra.CassandraResourceManager;
 import org.apache.beam.it.common.PipelineLauncher;
 import org.apache.beam.it.common.PipelineOperator;
@@ -102,14 +103,14 @@ public void setUp() throws Exception {
                 pubsubResourceManager,
                 getGcsPath("dlq", gcsResourceManager).replace("gs://" + artifactBucketName, ""),
                 gcsResourceManager);
-        ADDITIONAL_JOB_PARAMS.putAll(
+        Map<String, String> jobParameters =
             new HashMap<>() {
               {
                 put("network", VPC_NAME);
                 put("subnetwork", SUBNET_NAME);
                 put("workerRegion", VPC_REGION);
               }
-            });
+            };
         jobInfo =
             launchDataflowJob(
                 gcsResourceManager,
@@ -121,7 +122,8 @@ public void setUp() throws Exception {
                 null,
                 null,
                 null,
-                CASSANDRA_SOURCE_TYPE);
+                CASSANDRA_SOURCE_TYPE,
+                jobParameters);
       }
     }
   }