Fix GcsConnectorUtil auth for gcs-connector 3.x (#5984)

Author: shnapz

scio-parquet/src/main/scala/com/spotify/scio/parquet/GcsConnectorUtil.scala

@@ -48,19 +48,23 @@ private[parquet] object GcsConnectorUtil {
       case _                            => None
     } match {
       case Success(Some(sa)) =>
+        conf.set("fs.gs.auth.type", "SERVICE_ACCOUNT_JSON_KEYFILE")
         conf.set("fs.gs.auth.service.account.json.keyfile", sa)
       case Success(None) =>
+        conf.set("fs.gs.auth.type", "ACCESS_TOKEN_PROVIDER")
         conf.set(
           "fs.gs.auth.access.token.provider.impl",
           "com.spotify.scio.parquet.ApplicationDefaultTokenProvider"
         )
       case _ =>
+        conf.set("fs.gs.auth.type", "UNAUTHENTICATED")
         conf.setBoolean("fs.gs.auth.service.account.enable", false)
         conf.setBoolean("fs.gs.auth.null.enable", true)
     }
   }
 
   def unsetCredentials(conf: Configuration): Unit = {
+    conf.unset("fs.gs.auth.type")
     conf.unset("fs.gs.auth.service.account.json.keyfile")
     conf.unset("fs.gs.auth.access.token.provider.impl")
     conf.unset("fs.gs.auth.null.enable")

scio-parquet/src/test/scala/com/spotify/scio/parquet/GcsConnectorUtilTest.scala

@@ -25,6 +25,26 @@ import org.scalatest.matchers.should.Matchers
 
 class GcsConnectorUtilTest extends AnyFlatSpec with Matchers {
 
+  "setCredentials" should "set SERVICE_ACCOUNT_JSON_KEYFILE auth type for service account credentials" in {
+    val conf = ParquetConfiguration.empty()
+    sys.env.get("GOOGLE_APPLICATION_CREDENTIALS").foreach { _ =>
+      GcsConnectorUtil.setCredentials(conf)
+      val authType = conf.get("fs.gs.auth.type")
+      authType should (
+        equal("SERVICE_ACCOUNT_JSON_KEYFILE") or equal("ACCESS_TOKEN_PROVIDER")
+      )
+    }
+  }
+
+  it should "set auth type and clear it on unset" in {
+    val conf = ParquetConfiguration.empty()
+    GcsConnectorUtil.setCredentials(conf)
+    conf.get("fs.gs.auth.type") should not be null
+
+    GcsConnectorUtil.unsetCredentials(conf)
+    conf.get("fs.gs.auth.type") shouldBe null
+  }
+
   private def getInputPaths(sc: ScioContext, path: String): Array[String] = {
     val conf = ParquetConfiguration.empty()
     GcsConnectorUtil.setInputPaths(sc, conf, path)