Fix BT integration tests (#5606)

Author: kellen

integration/src/test/scala/com/spotify/scio/bigtable/BigtableIT.scala

@@ -110,7 +110,7 @@ class BigtableIT extends PipelineSpec {
         .build()
       runWithRealContext() { sc =>
         sc
-          .bigtable(projectId, instanceId, tableId, rowFilter = rowFilter)
+          .bigtable(BTOptions(projectId, instanceId), tableId, rowFilter = rowFilter)
           .map(fromRow) should containInAnyOrder(data)
       }.waitUntilDone()
     } catch {
@@ -151,7 +151,7 @@ class BigtableIT extends PipelineSpec {
         .build()
       runWithRealContext() { sc =>
         sc
-          .bigtable(projectId, instanceId, tableId, rowFilter = rowFilter)
+          .bigtable(BTOptions(projectId, instanceId), tableId, rowFilter = rowFilter)
           .map(fromRow) should containInAnyOrder(data)
       }.waitUntilDone()
     } catch {

scio-google-cloud-platform/src/main/scala/com/spotify/scio/bigtable/syntax/SCollectionSyntax.scala

@@ -23,7 +23,7 @@ import com.google.protobuf.ByteString
 import com.spotify.scio.io.ClosedTap
 import com.spotify.scio.values.SCollection
 import org.joda.time.Duration
-import com.spotify.scio.bigtable.{BigtableTypedIO, BigtableWrite}
+import com.spotify.scio.bigtable.{BTOptions, BigtableTypedIO, BigtableWrite}
 import com.spotify.scio.coders.Coder
 import magnolify.bigtable.BigtableType
 
@@ -58,8 +58,6 @@ final class SCollectionMutationOps[T <: Mutation](
 final class BigtableTypedOps[K: Coder, T: BigtableType: Coder](
   private val self: SCollection[(K, T)]
 ) {
-  private def btOpts(projectId: String, instanceId: String): BigtableOptions =
-    BigtableOptions.builder().setProjectId(projectId).setInstanceId(instanceId).build
 
   def saveAsBigtable(
     projectId: String,
@@ -69,7 +67,7 @@ final class BigtableTypedOps[K: Coder, T: BigtableType: Coder](
     keyFn: K => ByteString
   ): ClosedTap[Nothing] = {
     val params = BigtableTypedIO.WriteParam[K](columnFamily, keyFn)
-    self.write(BigtableTypedIO[K, T](btOpts(projectId, instanceId), tableId))(params)
+    self.write(BigtableTypedIO[K, T](BTOptions(projectId, instanceId), tableId))(params)
   }
 
   def saveAsBigtable(
@@ -81,7 +79,7 @@ final class BigtableTypedOps[K: Coder, T: BigtableType: Coder](
     timestamp: Long
   ): ClosedTap[Nothing] = {
     val params = BigtableTypedIO.WriteParam[K](columnFamily, keyFn, timestamp)
-    self.write(BigtableTypedIO[K, T](btOpts(projectId, instanceId), tableId))(params)
+    self.write(BigtableTypedIO[K, T](BTOptions(projectId, instanceId), tableId))(params)
   }
 
   def saveAsBigtable(

scio-google-cloud-platform/src/main/scala/com/spotify/scio/bigtable/syntax/ScioContextSyntax.scala

@@ -22,7 +22,7 @@ import com.google.bigtable.v2._
 import com.google.cloud.bigtable.config.BigtableOptions
 import com.google.protobuf.ByteString
 import com.spotify.scio.ScioContext
-import com.spotify.scio.bigtable.{BigtableRead, BigtableTypedIO, BigtableUtil, TableAdmin}
+import com.spotify.scio.bigtable.{BTOptions, BigtableRead, BigtableTypedIO, BigtableUtil, TableAdmin}
 import com.spotify.scio.coders.Coder
 import com.spotify.scio.values.SCollection
 import magnolify.bigtable.BigtableType
@@ -40,17 +40,14 @@ object ScioContextOps {
 final class ScioContextOps(private val self: ScioContext) extends AnyVal {
   import ScioContextOps._
 
-  private def btOpts(projectId: String, instanceId: String): BigtableOptions =
-    BigtableOptions.builder().setProjectId(projectId).setInstanceId(instanceId).build
-
   def typedBigtable[K: Coder, T: BigtableType: Coder](
     projectId: String,
     instanceId: String,
     tableId: String,
     columnFamily: String,
     keyFn: ByteString => K
   ): SCollection[(K, T)] =
-    typedBigtable(btOpts(projectId, instanceId), tableId, columnFamily, keyFn)
+    typedBigtable(BTOptions(projectId, instanceId), tableId, columnFamily, keyFn)
 
   def typedBigtable[K: Coder, T: BigtableType: Coder](
     projectId: String,
@@ -60,7 +57,7 @@ final class ScioContextOps(private val self: ScioContext) extends AnyVal {
     keyFn: ByteString => K,
     keyRanges: Seq[ByteKeyRange]
   ): SCollection[(K, T)] =
-    typedBigtable(btOpts(projectId, instanceId), tableId, columnFamily, keyFn, keyRanges)
+    typedBigtable(BTOptions(projectId, instanceId), tableId, columnFamily, keyFn, keyRanges)
 
   def typedBigtable[K: Coder, T: BigtableType: Coder](
     projectId: String,
@@ -71,7 +68,7 @@ final class ScioContextOps(private val self: ScioContext) extends AnyVal {
     keyRanges: Seq[ByteKeyRange],
     rowFilter: RowFilter
   ): SCollection[(K, T)] =
-    typedBigtable(btOpts(projectId, instanceId), tableId, columnFamily, keyFn, keyRanges, rowFilter)
+    typedBigtable(BTOptions(projectId, instanceId), tableId, columnFamily, keyFn, keyRanges, rowFilter)
 
   def typedBigtable[K: Coder, T: BigtableType: Coder](
     projectId: String,
@@ -84,7 +81,7 @@ final class ScioContextOps(private val self: ScioContext) extends AnyVal {
     maxBufferElementCount: Option[Int]
   ): SCollection[(K, T)] =
     typedBigtable(
-      btOpts(projectId, instanceId),
+      BTOptions(projectId, instanceId),
       tableId,
       columnFamily,
       keyFn,
@@ -115,7 +112,7 @@ final class ScioContextOps(private val self: ScioContext) extends AnyVal {
     keyRange: ByteKeyRange,
     rowFilter: RowFilter
   ): SCollection[Row] =
-    bigtable(btOpts(projectId, instanceId), tableId, Seq(keyRange), rowFilter)
+    bigtable(BTOptions(projectId, instanceId), tableId, Seq(keyRange), rowFilter)
 
   /** Get an SCollection for a Bigtable table. */
   def bigtable(
@@ -127,7 +124,7 @@ final class ScioContextOps(private val self: ScioContext) extends AnyVal {
     maxBufferElementCount: Option[Int]
   ): SCollection[Row] =
     bigtable(
-      btOpts(projectId, instanceId),
+      BTOptions(projectId, instanceId),
       tableId,
       Seq(keyRange),
       rowFilter,
@@ -140,7 +137,7 @@ final class ScioContextOps(private val self: ScioContext) extends AnyVal {
     instanceId: String,
     tableId: String
   ): SCollection[Row] =
-    bigtable(btOpts(projectId, instanceId), tableId)
+    bigtable(BTOptions(projectId, instanceId), tableId)
 
   /** Get an SCollection for a Bigtable table. */
   def bigtable(
@@ -151,7 +148,7 @@ final class ScioContextOps(private val self: ScioContext) extends AnyVal {
     rowFilter: RowFilter,
     maxBufferElementCount: Option[Int]
   ): SCollection[Row] =
-    bigtable(btOpts(projectId, instanceId), tableId, keyRanges, rowFilter, maxBufferElementCount)
+    bigtable(BTOptions(projectId, instanceId), tableId, keyRanges, rowFilter, maxBufferElementCount)
 
   /** Get an SCollection for a Bigtable table. */
   def bigtable(