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
TRAINING_DEFAULT_ARGS = {
        'project_id': 'test-project',
        'job_id': 'test_training',
        'task_id': 'test-training',
    }
    @patch('airflow.providers.google.cloud.operators.mlengine.MLEngineHook')
    def test_success_cancel_training_job(self, mock_hook):
        success_response = {}
        hook_instance = mock_hook.return_value
        hook_instance.cancel_job.return_value = success_response
        cancel_training_op = MLEngineTrainingCancelJobOperator(**self.TRAINING_DEFAULT_ARGS)
        cancel_training_op.execute(None)
        mock_hook.assert_called_once_with(
            gcp_conn_id='google_cloud_default',
            delegate_to=None,
            impersonation_chain=None,
        )
        
        self.assertEqual(len(hook_instance.mock_calls), 1)
        hook_instance.cancel_job.assert_called_once_with(
            project_id=self.TRAINING_DEFAULT_ARGS['project_id'], job_id=self.TRAINING_DEFAULT_ARGS['job_id']
        )
    @patch('airflow.providers.google.cloud.operators.mlengine.MLEngineHook')
    def test_http_error(self, mock_hook):