[Issue-154] Update connector version to 0.13.0-SNAPSHOT

README.md

@@ -39,13 +39,12 @@ The [master](https://github.com/pravega/spark-connectors) branch will always hav
 
 | Spark Version | Pravega Version | Java Version To Build Connector | Java Version To Run Connector | Git Branch                                                                        |
 |---------------|-----------------|---------------------------------|-------------------------------|-----------------------------------------------------------------------------------|
-| 3.1+          | 0.12            | Java 11                         | Java 8 or 11                  | [master](https://github.com/pravega/spark-connectors)                             |
+| 3.1+          | 0.13            | Java 11                         | Java 8 or 11                  | [master](https://github.com/pravega/spark-connectors)                             |
+| 2.4           | 0.13            | Java 8                          | Java 8                        | [r0.13-spark2.4](https://github.com/pravega/spark-connectors/tree/r0.13-spark2.4) |
+| 3.1+          | 0.12            | Java 11                         | Java 8 or 11                  | [r0.12](https://github.com/pravega/spark-connectors/tree/r0.12)                   |
 | 2.4           | 0.12            | Java 8                          | Java 8                        | [r0.12-spark2.4](https://github.com/pravega/spark-connectors/tree/r0.12-spark2.4) |
 | 3.1+          | 0.11            | Java 11                         | Java 8 or 11                  | [r0.11](https://github.com/pravega/spark-connectors/tree/r0.11)                   |
 | 2.4           | 0.11            | Java 8                          | Java 8                        | [r0.11-spark2.4](https://github.com/pravega/spark-connectors/tree/r0.11-spark2.4) |
-| 3.1+          | 0.10            | Java 11                         | Java 8 or 11                  | [r0.10](https://github.com/pravega/spark-connectors/tree/r0.10)                   |
-| 3.0           | 0.10            | Java 11                         | Java 8 or 11                  | [r0.10-spark3.0](https://github.com/pravega/spark-connectors/tree/r0.10-spark3.0) |
-| 2.4           | 0.10            | Java 8                          | Java 8                        | [r0.10-spark2.4](https://github.com/pravega/spark-connectors/tree/r0.10-spark2.4) |
 
 ## Support
 

gradle.properties

@@ -26,8 +26,8 @@ slf4jApiVersion=1.7.25
 sparkVersion=3.3.0
 
 # Version and base tags can be overridden at build time.
-connectorVersion=0.12.0-SNAPSHOT
-pravegaVersion=0.12.0-3099.d57bc8b-SNAPSHOT
+connectorVersion=0.13.0-SNAPSHOT
+pravegaVersion=0.13.0-3142.2542966-SNAPSHOT
 
 # flag to indicate if Pravega sub-module should be used instead of the version defined in 'pravegaVersion'
 usePravegaVersionSubModule=false

src/main/scala/io/pravega/connectors/spark/PravegaMetaScan.scala

@@ -25,7 +25,7 @@ class PravegaMetaScan(options: CaseInsensitiveStringMap, metadataTableName: Meta
     (for (streamManager <- managed(StreamManager.create(clientConfig))) yield {
       metadataTableName match {
         case MetadataTableName.StreamInfo => {
-          val streamInfo = streamManager.getStreamInfo(scopeName, streamName)
+          val streamInfo = PravegaUtils.getStreamInfo(streamManager, scopeName, streamName)
           val schema = StructType(Seq(
             StructField("head_stream_cut", StringType),
             StructField("tail_stream_cut", StringType)

src/main/scala/io/pravega/connectors/spark/PravegaMicroBatchStream.scala

@@ -55,7 +55,7 @@ class PravegaMicroBatchStream(
   // Resolve start and end stream cuts now.
   // We must ensure that getStartOffset and getEndOffset return specific stream cuts, even
   // if the caller passes "earliest" or "latest".
-  private lazy val initialStreamInfo = streamManager.getStreamInfo(scopeName, streamName)
+  private lazy val initialStreamInfo = PravegaUtils.getStreamInfo(streamManager, scopeName, streamName)
 
   private val resolvedStartStreamCut = startStreamCut match {
     case EarliestStreamCut | UnboundedStreamCut => initialStreamInfo.getHeadStreamCut
@@ -76,7 +76,7 @@ class PravegaMicroBatchStream(
   }
 
   override def latestOffset(): Offset = {
-    PravegaSourceOffset(streamManager.getStreamInfo(scopeName, streamName).getTailStreamCut)
+    PravegaSourceOffset(PravegaUtils.getStreamInfo(streamManager, scopeName, streamName).getTailStreamCut)
   }
 
   /**
@@ -93,7 +93,7 @@ class PravegaMicroBatchStream(
    */
   override def planInputPartitions(start: Offset, end: Offset): Array[InputPartition] = {
     log.info(s"planInputPartitions(${start},${end})")
-    lazy val streamInfo = streamManager.getStreamInfo(scopeName, streamName)
+    lazy val streamInfo = PravegaUtils.getStreamInfo(streamManager, scopeName, streamName)
     batchStartStreamCut = Option(start)
       .map(_.asInstanceOf[PravegaSourceOffset].streamCut)
       .getOrElse(resolvedStartStreamCut)

src/main/scala/io/pravega/connectors/spark/PravegaUtils.scala

@@ -0,0 +1,18 @@
+/**
+ * Copyright (c) Dell Inc., or its subsidiaries. All Rights Reserved.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ */
+package io.pravega.connectors.spark
+
+import io.pravega.client.admin.{StreamInfo, StreamManager}
+
+private object PravegaUtils {
+    def getStreamInfo(streamManager: StreamManager, scopeName: String, streamName: String): StreamInfo = {
+        streamManager.fetchStreamInfo(scopeName, streamName).join()
+    }
+}

src/test/scala/io/pravega/connectors/spark/PravegaTestUtils.scala

@@ -96,7 +96,7 @@ class PravegaTestUtils extends Logging {
   def getStreamInfo(streamName: String): StreamInfo = {
     val streamManager = StreamManager.create(SETUP_UTILS.getControllerUri)
     try {
-      streamManager.getStreamInfo(scope, streamName)
+      PravegaUtils.getStreamInfo(streamManager, scope, streamName)
     } finally {
       streamManager.close()
     }