Fix Tests

Author: clun

src/test/java/org/apache/beam/sdk/io/astra/AbstractAstraTest.java

@@ -95,7 +95,7 @@ protected static byte[] getSecureBundle(String dbname, String keyspace) {
      *      Cassandra cluster
      */
     protected static CqlSession getCqlSession(String dbName, String keyspace) {
-        if (cqlSession == null) {
+        if (cqlSession == null || cqlSession.isClosed()) {
             cqlSession = CqlSession.builder()
                     .withCloudSecureConnectBundle(getSecureConnectBundlePath(dbName, keyspace))
                     .withAuthCredentials("token", getToken())
@@ -159,7 +159,8 @@ protected static AstraDevopsApiClient getApiDevopsClient() {
      * @return
      *      database client
      */
-    protected static DatabaseClient createDbAndProvideClient(String dbName, String keyspace, boolean vector) throws InterruptedException {
+    protected static DatabaseClient createDbAndProvideClient(String dbName, String keyspace, boolean vector)
+    throws InterruptedException {
         if (dbClient == null) {
             if (!getDatabasesClient().findByName(dbName).findAny().isPresent()) {
                 LOGGER.info("Create DB  {} as not existing ", dbName);

src/test/java/org/apache/beam/sdk/io/astra/AstraDbIOVectorTest.java

@@ -43,10 +43,10 @@ public class AstraDbIOVectorTest extends AbstractAstraTest implements Serializab
 
     @BeforeClass
     public static void beforeClass() throws InterruptedException {
-        //createDbAndProvideClient(DB_VECTOR_NAME, DB_KEYSPACE_NAME, true);
+        createDbAndProvideClient(DB_VECTOR_NAME, DB_KEYSPACE_NAME, true);
         cqlSession = getCqlSession(DB_VECTOR_NAME, DB_KEYSPACE_NAME);
-        //createSchema();
-        //insertData();
+        createSchema();
+        insertData();
     }
 
     @Test
@@ -152,11 +152,7 @@ public void processElement(ProcessContext c) {
             Arrays.asList(vector).stream().map(f -> df.format(f))
                     .collect(Collectors.toList()).toString();
             System.out.println("Decoded Vector: " + result);
-
-             c.output(c.element().toString());
-            //CqlVectorCodec codec = new CqlVectorCodec<Float>();
-
-
+            c.output(c.element().toString());
         }
     }