Cassandra Reverse Replication - Integration Tests

v2/spanner-to-sourcedb/pom.xml

@@ -106,6 +106,17 @@
             <version>${project.version}</version>
             <scope>test</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.beam</groupId>
+            <artifactId>beam-it-cassandra</artifactId>
+            <scope>test</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>com.datastax.cassandra</groupId>
+                    <artifactId>cassandra-driver-core</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
     </dependencies>
 
 </project>

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

@@ -121,6 +121,11 @@ private interface HandlerSupplier<T> {
   private static BigInteger handleCassandraVarintType(Object value) {
     if (value instanceof byte[]) {
       return new BigInteger((byte[]) value);
+    } else if (value instanceof ByteBuffer) {
+      ByteBuffer byteBuffer = (ByteBuffer) value;
+      byte[] byteArray = new byte[byteBuffer.remaining()];
+      byteBuffer.get(byteArray);
+      return new BigInteger(byteArray);
     }
     return new BigInteger(value.toString());
   }
@@ -166,17 +171,12 @@ private static InetAddress handleCassandraInetAddressType(String inetString) {
    * @return a {@link ByteBuffer} object containing the value represented in cassandra type.
    */
   private static ByteBuffer parseBlobType(Object colValue) {
-    byte[] byteArray;
-
     if (colValue instanceof byte[]) {
-      byteArray = (byte[]) colValue;
-    } else if (colValue instanceof String) {
-      byteArray = java.util.Base64.getDecoder().decode((String) colValue);
-    } else {
-      throw new IllegalArgumentException("Unsupported type for column");
+      return ByteBuffer.wrap((byte[]) colValue);
+    } else if (colValue instanceof ByteBuffer) {
+      return (ByteBuffer) colValue;
     }
-
-    return ByteBuffer.wrap(byteArray);
+    return ByteBuffer.wrap(java.util.Base64.getDecoder().decode((String) colValue));
   }
 
   /**
@@ -331,7 +331,14 @@ private static Object handleSpannerColumnType(
           return null;
         }
         String hexEncodedString = valuesJson.optString(columnName);
-        return convertBinaryEncodedStringToByteArray(hexEncodedString);
+        return safeHandle(
+            () -> {
+              try {
+                return safeHandle(() -> convertBinaryEncodedStringToByteArray(hexEncodedString));
+              } catch (IllegalArgumentException e) {
+                return parseBlobType(hexEncodedString);
+              }
+            });
       } else {
         return valuesJson.isNull(columnName) ? null : valuesJson.opt(columnName);
       }