@@ -386,6 +386,12 @@ def list_data_sources(
    def apply_data_source(
        self, data_source: DataSource, project: str, commit: bool = True
    ):
        now = _utc_now()
        if not data_source.created_timestamp:
            data_source.created_timestamp = now
@@ -394,7 +400,10 @@ def apply_data_source(
        registry = self._prepare_registry_for_changes(project)

        for idx, existing_data_source_proto in enumerate(registry.data_sources):
-            if existing_data_source_proto.name == data_source.name:
                existing_data_source = DataSource.from_proto(existing_data_source_proto)
                # Check if the data source has actually changed
                if existing_data_source == data_source:
@@ -423,7 +432,7 @@ def delete_data_source(self, name: str, project: str, commit: bool = True):
        for idx, data_source_proto in enumerate(
            self.cached_registry_proto.data_sources
        ):
-            if data_source_proto.name == name:
                del self.cached_registry_proto.data_sources[idx]
                if commit:
                    self.commit()

@@ -899,6 +899,111 @@ def test_apply_data_source_with_timestamps(test_registry):
    test_registry.teardown()


@pytest.mark.integration
@pytest.mark.parametrize(
    "test_registry",