Training with a custom service account needs a new field named service account
trainingInput:
  serviceAccount: CUSTOM_SERVICE_ACCOUNT

https://cloud.google.com/ai-platform/training/docs/custom-service-account


airflow/providers/google/cloud/operators/mlengine.py
job_id = _normalize_mlengine_job_id(self._job_id)
        prediction_request = {
            'jobId': job_id,
            'predictionInput': {
                'dataFormat': self._data_format,
                'inputPaths': self._input_paths,
                'outputPath': self._output_path,
                'region': self._region,
            },
        }
        if self._labels:
            prediction_request['labels'] = self._labels
        if self._uri:
            prediction_request['predictionInput']['uri'] = self._uri
        elif self._model_name:
            origin_name = 'projects/{}/models/{}'.format(self._project_id, self._model_name)
            if not self._version_name:
                prediction_request['predictionInput']['modelName'] = origin_name
            else:
                prediction_request['predictionInput']['versionName'] = origin_name + '/versions/{}'.format(
                    self._version_name
                )
        if self._max_worker_count:
            prediction_request['predictionInput']['maxWorkerCount'] = self._max_worker_count
        if self._runtime_version: