How can authenticate in dataflow + Apache beam using Java or Kotlin if I have the content of the service account in the variable?

I tried this code but It doesn't work. However, the service account is correct.

val dataflowOptions = PipelineOptionsFactory.create().`as`(MyOptions::class.java)
       dataflowOptions.jobName = "tesjob1"
       dataflowOptions.project = getGcpProjectId()
       dataflowOptions.region = getGcpRegion()
       dataflowOptions.subnetwork = getGcpSubNetwork()
       dataflowOptions.usePublicIps =false
       dataflowOptions.runner = DataflowRunner::class.java

val serviceAccountJson= ""

val credentials: Credentials = GoogleCredentials.fromStream(ByteArrayInputStream(serviceAccountJson.toByteArray(Charsets.UTF_8)))
dataflowOptions.gcpCredential = credentials

dataflowOptions.setGcpTempLocation("gs://deib_bucket/temp")
return Pipeline.create(dataflowOptions)

Could you give me any suggestions to resolve the issue?

I resolved the issue, I couldn't use the service account because I need to pass the scope, for that I changed the code and now it worked properly.

 val gcpSCOPES: List<String> = listOf(
        "https://www.googleapis.com/auth/cloud-platform",
        "https://www.googleapis.com/auth/devstorage.full_control",
        "https://www.googleapis.com/auth/userinfo.email",
        "https://www.googleapis.com/auth/datastore",
        "https://www.googleapis.com/auth/pubsub"
    )

        val options: DataflowPipelineOptions = PipelineOptionsFactory.create().`as`(DataflowPipelineOptions::class.java)
        options.jobName = "jobExample"
        options.project = getGcpProjectId()
        options.region = getGcpRegion()
        options.subnetwork = getGcpSubNetwork()
        options.usePublicIps =false
        options.runner = DataflowRunner::class.java
        options.gcpCredential = ServiceAccountCredentials.fromStream(ByteArrayInputStream(DeibDataSyncEnv.getGcpCredentials().toByteArray(Charsets.UTF_8)))
            .createScoped(gcpSCOPES)
        options.serviceAccount = getGcpServiceAccountEmail()

options.setGcpTempLocation(DeibDataSyncEnv.getGcpTempLocation())
        return Pipeline.create(dataflowOptions)

To help you and give more explanations on Dataflow job authentication and the use of Service Account :

If you launch your Dataflow job from your local machine, your have to export the GOOGLE_APPLICATION_CREDENTIALS env var, unfortunately there is no choice in this case.

However if you launch your job via a DAG orchestrator like Airflow and Cloud Composer no need to pass a SA key file. The authentication is handled with Airflow and the SA used by Cloud Composer.

You can also explore other solutions with Cloud Shell or Cloud Build to launch your job, but I think it's better for the CI CD part to deploy the job and delegate the responsability of job execution to a pipeline orchestration tool like Airflow.

In production environment, you can also use Dataflow Flex Template to standardize the deployment of your Dataflow jobs based on a bucket and a Docker image.

In this case, if you use a tool like Cloud Build, no need to pass a Service Account key file.

You can check this article I written, that shows a complete example of CI CD pipeline with Dataflow Flex Template.

In prod and to have the best standardisation for your jobs, I think Flex template should be the best solution.

An example of command to deploy a job with Flex Template :

gcloud dataflow flex-template build "$METADATA_TEMPLATE_FILE_PATH" \
  --image-gcr-path "$LOCATION-docker.pkg.dev/$PROJECT_ID/$REPO_NAME/$IMAGE_NAME:$IMAGE_TAG" \
  --sdk-language "$SDK_LANGUAGE" \
  --flex-template-base-image "$FLEX_TEMPLATE_BASE_IMAGE" \
  --metadata-file "$METADATA_FILE" \
  --jar "$JAR" \
  --env FLEX_TEMPLATE_JAVA_MAIN_CLASS="$FLEX_TEMPLATE_JAVA_MAIN_CLASS"

An example of command to launch the job with the serviceAccount program argument :

gcloud dataflow flex-template run "$JOB_NAME-$(date +%Y%m%d-%H%M%S)" \
  --template-file-gcs-location "$METADATA_TEMPLATE_FILE_PATH" \
  --project="$PROJECT_ID" \
  --region="$LOCATION" \
  --temp-location="$TEMP_LOCATION" \
  --staging-location="$STAGING_LOCATION" \
  --parameters serviceAccount="$SA_EMAIL" \
  --parameters inputJsonFile="$INPUT_FILE" \
  --parameters inputFileSlogans="$SIDE_INPUT_FILE" \
  --parameters teamLeagueDataset="$TEAM_LEAGUE_DATASET" \
  --parameters teamStatsTable="$TEAM_STATS_TABLE" \
  --parameters jobType="$JOB_TYPE" \
  --parameters failureOutputDataset="$FAILURE_OUTPUT_DATASET" \
  --parameters failureOutputTable="$FAILURE_OUTPUT_TABLE" \
  --parameters failureFeatureName="$FAILURE_FEATURE_NAME"