Fixing offset for Spark 2 -> Spark 3 offset conversion (Azure#33757)

Author: FabianMeiswinkel

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/implementation/SparkBridgeImplementationInternal.scala

@@ -277,7 +277,7 @@ private[cosmos] object SparkBridgeImplementationInternal extends BasicLoggingTra
 
         // Spark 3 tracks the last LSN for which docs have been successfully processed
         // Spark 2 LSN is offset by 1 because in Spark 2 the next-to-be-sent-as-continuation LSN is tracked
-        val lsn: Long = token._2 + 1
+        val lsn: Long = token._2 - 1
 
         val range: Range[String] = if (pkRangesByPkRangeId.contains(pkRangeId)) {
           pkRangesByPkRangeId.get(pkRangeId).get.toRange

sdk/cosmos/azure-cosmos-spark_3_2-12/src/main/scala/com/azure/cosmos/spark/udf/CreateSpark2ContinuationsFromChangeFeedOffset.scala

@@ -112,7 +112,7 @@ class CreateSpark2ContinuationsFromChangeFeedOffset extends UDF2[Map[String, Str
             pkRange.getId.toInt,
             // Spark 3 tracks the last LSN for which docs have been successfully processed
             // Spark 2 LSN is offset by 1 because in Spark 2 the next-to-be-sent-as-continuation LSN is tracked
-            Math.max(0, SparkBridgeImplementationInternal.toLsn(minLsn.get.getToken) - 1))
+            Math.max(0, SparkBridgeImplementationInternal.toLsn(minLsn.get.getToken) + 1))
         }
       })
 

sdk/cosmos/azure-cosmos-spark_3_2-12/src/test/scala/com/azure/cosmos/spark/SparkE2EChangeFeedITest.scala

@@ -730,7 +730,7 @@ class SparkE2EChangeFeedITest
             val quotedToken = filteredCompositeContinuations.head.getToken
             // Spark 3 tracks the last LSN for which docs have been successfully processed
             // Spark 2 LSN is offset by 1 because in Spark 2 the next-to-be-sent-as-continuation LSN is tracked
-            val lsn: Long = Math.max(0, quotedToken.substring(1, quotedToken.length - 1).toLong - 1)
+            val lsn: Long = Math.max(0, quotedToken.substring(1, quotedToken.length - 1).toLong + 1)
 
             tokenMap += (pkRangeId -> lsn)
           })