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
if not self._model_name:
            raise AirflowException("The model_name parameter could not be empty.")
        if not self._version:
            raise AirflowException("The version parameter could not be empty.")
    def execute(self, context):
        hook = MLEngineHook(
            gcp_conn_id=self._gcp_conn_id,
            delegate_to=self._delegate_to,
            impersonation_chain=self._impersonation_chain,
        )
        return hook.create_version(
            project_id=self._project_id, model_name=self._model_name, version_spec=self._version
        )
class MLEngineSetDefaultVersionOperator(BaseOperator):