Fix asset loader for sourceassets

Author: dpeng817

python_modules/dagster/dagster/_core/definitions/module_loaders/object_list.py

@@ -349,11 +349,7 @@ def with_attributes(
                     else new_asset
                 )
             elif isinstance(dagster_def, SourceAsset):
-                return_list.append(
-                    dagster_def.with_attributes(
-                        group_name=group_name if group_name else dagster_def.group_name
-                    )
-                )
+                return_list.append(dagster_def.with_attributes(group_name=group_name))
             elif isinstance(dagster_def, AssetSpec):
                 return_list.append(
                     _spec_mapper_disallow_group_override(group_name, automation_condition)(

python_modules/dagster/dagster/_core/definitions/source_asset.py

@@ -460,9 +460,13 @@ def with_resources(self, resource_defs) -> "SourceAsset":
     def with_attributes(
         self, group_name: Optional[str] = None, key: Optional[AssetKey] = None
     ) -> "SourceAsset":
-        if group_name is not None and self.group_name != DEFAULT_GROUP_NAME:
+        if (
+            group_name is not None
+            and self.group_name != DEFAULT_GROUP_NAME
+            and self.group_name != group_name
+        ):
             raise DagsterInvalidDefinitionError(
-                "A group name has already been provided to source asset"
+                f"Attempted to override group name to {group_name} for SourceAsset {self.key.to_user_string()}, which already has group name {self.group_name}."
                 f" {self.key.to_user_string()}"
             )