OmegaConfigLoader should not show MissingConfigException when the file is empty

RELEASE.md

@@ -15,6 +15,7 @@
 
 ## Bug fixes and other changes
 * `OmegaConfigLoader` will return a `dict` instead of `DictConfig`.
+* `OmegaConfigLoader` does not show a `MissingConfigError` when the config files exist but are empty.
 
 ## Breaking changes to the API
 * `kedro package` does not produce `.egg` files anymore, and now relies exclusively on `.whl` files.

kedro/config/omegaconf_config.py

@@ -162,13 +162,14 @@ def __getitem__(self, key) -> dict[str, Any]:
 
         read_environment_variables = key == "credentials"
 
+        processed_files: set[Path] = set()
         # Load base env config
         if self._protocol == "file":
             base_path = str(Path(self.conf_source) / self.base_env)
         else:
             base_path = str(Path(self._fs.ls("", detail=False)[-1]) / self.base_env)
         base_config = self.load_and_merge_dir_config(
-            base_path, patterns, key, read_environment_variables
+            base_path, patterns, key, processed_files, read_environment_variables
         )
         config = base_config
 
@@ -179,9 +180,8 @@ def __getitem__(self, key) -> dict[str, Any]:
         else:
             env_path = str(Path(self._fs.ls("", detail=False)[-1]) / run_env)
         env_config = self.load_and_merge_dir_config(
-            env_path, patterns, key, read_environment_variables
+            env_path, patterns, key, processed_files, read_environment_variables
         )
-
         # Destructively merge the two env dirs. The chosen env will override base.
         common_keys = config.keys() & env_config.keys()
         if common_keys:
@@ -194,7 +194,7 @@ def __getitem__(self, key) -> dict[str, Any]:
 
         config.update(env_config)
 
-        if not config:
+        if not processed_files:
             raise MissingConfigException(
                 f"No files of YAML or JSON format found in {base_path} or {env_path} matching"
                 f" the glob pattern(s): {[*self.config_patterns[key]]}"
@@ -207,11 +207,12 @@ def __repr__(self):  # pragma: no cover
             f"config_patterns={self.config_patterns})"
         )
 
-    def load_and_merge_dir_config(
+    def load_and_merge_dir_config(  # pylint: disable=too-many-arguments
         self,
         conf_path: str,
         patterns: Iterable[str],
         key: str,
+        processed_files: set,
         read_environment_variables: bool | None = False,
     ) -> dict[str, Any]:
         """Recursively load and merge all configuration files in a directory using OmegaConf,
@@ -221,6 +222,7 @@ def load_and_merge_dir_config(
             conf_path: Path to configuration directory.
             patterns: List of glob patterns to match the filenames against.
             key: Key of the configuration type to fetch.
+            processed_files: Set of files read for a given configuration type.
             read_environment_variables: Whether to resolve environment variables.
 
         Raises:
@@ -258,6 +260,7 @@ def load_and_merge_dir_config(
                     # this is a workaround to read it as a binary file and decode it back to utf8.
                     tmp_fo = io.StringIO(open_config.read().decode("utf8"))
                     config = OmegaConf.load(tmp_fo)
+                    processed_files.add(config_filepath)
                 if read_environment_variables:
                     self._resolve_environment_variables(config)
                 config_per_file[config_filepath] = config

tests/config/test_omegaconf_config.py

@@ -335,6 +335,15 @@ def test_no_files_found(self, tmp_path):
         with pytest.raises(MissingConfigException, match=pattern):
             OmegaConfigLoader(str(tmp_path))["credentials"]
 
+    def test_empty_catalog_file(self, tmp_path):
+        """Check that empty catalog file is read and returns an empty dict"""
+        _write_yaml(tmp_path / _BASE_ENV / "catalog_empty.yml", {})
+        catalog_patterns = {"catalog": ["catalog*", "catalog*/**", "**/catalog*"]}
+        catalog = OmegaConfigLoader(
+            conf_source=tmp_path, env="base", config_patterns=catalog_patterns
+        )["catalog"]
+        assert catalog == {}
+
     def test_overlapping_patterns(self, tmp_path, mocker):
         """Check that same configuration file is not loaded more than once."""
         _write_yaml(