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


tests/providers/google/cloud/operators/test_mlengine.py
)
        
        task_args = self.BATCH_PREDICTION_DEFAULT_ARGS.copy()
        task_args['version_name'] = 'bare_version'
        with self.assertRaises(AirflowException) as context:
            MLEngineStartBatchPredictionJobOperator(**task_args).execute(None)
        self.assertEqual(
            'Missing model: Batch prediction expects a model ' 'name when a version name is provided.',
            str(context.exception),
        )
        
        task_args = self.BATCH_PREDICTION_DEFAULT_ARGS.copy()
        with self.assertRaises(AirflowException) as context:
            MLEngineStartBatchPredictionJobOperator(**task_args).execute(None)
        self.assertEqual(
            'Missing model origin: Batch prediction expects a '
            'model, a model & version combination, or a URI to a savedModel.',
            str(context.exception),
        )
    @patch('airflow.providers.google.cloud.operators.mlengine.MLEngineHook')
    def test_http_error(self, mock_hook):
        http_error_code = 403
        input_with_model = self.INPUT_MISSING_ORIGIN.copy()
        input_with_model['modelName'] = 'projects/experimental/models/test_model'
        hook_instance = mock_hook.return_value
        hook_instance.create_job.side_effect = HttpError(
            resp=httplib2.Response({'status': http_error_code}), content=b'Forbidden'
        )
        with self.assertRaises(HttpError) as context:
            prediction_task = MLEngineStartBatchPredictionJobOperator(