This is an automated email from the ASF dual-hosted git repository.
jerryshao pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gravitino.git
The following commit(s) were added to refs/heads/main by this push:
new 6f83a812ec [MINOR] test: Wait for Hive Metastore readiness in Python
ITs (#13072)
6f83a812ec is described below
commit 6f83a812ecf70adf71c9d0eedd48db1df0f1773b
Author: Qi Yu <[email protected]>
AuthorDate: Fri Sep 11 17:17:08 2026 +0800
[MINOR] test: Wait for Hive Metastore readiness in Python ITs (#13072)
### What changes were proposed in this pull request?
Wait for a successful query through the Hive Metastore Thrift endpoint
before starting Python integration tests. Bound readiness commands and
polling with a deadline, and clean up the container if initialization
fails.
### Why are the changes needed?
The image's Hive CLI check can use an embedded metastore and report
success while port 9083 is unavailable. The previous asyncio timeout
cannot interrupt synchronous Docker calls and sleeps.
Observed in
[PythonIT](https://github.com/apache/gravitino/actions/runs/34436219502/job/102745773502),
where relational-table setup and catalog-tag cleanup failed with
Metastore connection refused.
### Does this PR introduce _any_ user-facing change?
No. Only Python integration-test infrastructure changes.
### How was this patch tested?
Four regression tests cover readiness retries, deadline handling, Docker
errors, and failed-startup cleanup. All passed for 50 consecutive runs
(200 executions, with mocked Docker and clock). Black, Pylint, and
spotlessApply passed. Full Docker PythonIT was not rerun locally.
CI repetition results for commit `958bcfc88d` (each complete PythonIT
attempt covers Python 3.10, 3.11, and 3.12):
- [Attempt
1](https://github.com/apache/gravitino/actions/runs/34444730076/job/102767005330):
success.
- [Attempt
2](https://github.com/apache/gravitino/actions/runs/34444730076/job/102779187802):
success.
- [Attempt
3](https://github.com/apache/gravitino/actions/runs/34444730076/job/102798135553):
success.
All three consecutive PythonIT attempts passed on the same commit.
---
.../tests/integration/containers/hdfs_container.py | 86 +++++++++++----------
.../tests/unittests/test_hdfs_container.py | 88 ++++++++++++++++++++++
2 files changed, 134 insertions(+), 40 deletions(-)
diff --git
a/clients/client-python/tests/integration/containers/hdfs_container.py
b/clients/client-python/tests/integration/containers/hdfs_container.py
index 3f34d65435..c4698c7e81 100644
--- a/clients/client-python/tests/integration/containers/hdfs_container.py
+++ b/clients/client-python/tests/integration/containers/hdfs_container.py
@@ -15,58 +15,60 @@
# specific language governing permissions and limitations
# under the License.
-import asyncio
import logging
import os
import time
from docker.errors import DockerException
+from docker.models.containers import Container
from gravitino.exceptions.base import GravitinoRuntimeException
-from gravitino.exceptions.base import InternalError
from tests.integration.containers.base_container import BaseContainer
logger = logging.getLogger(__name__)
-async def check_hdfs_status(hdfs_container):
- retry_limit = 15
- for _ in range(retry_limit):
+def check_hdfs_container_status(
+ hdfs_container: Container, timeout_sec: float = 150, interval_sec: float =
10
+) -> None:
+ """Wait for HDFS and the remote Hive Metastore, with bounded probe
commands."""
+ deadline = time.monotonic() + timeout_sec
+ last_output = b"No readiness probe completed"
+ while (remaining := deadline - time.monotonic()) > 0:
+ # The image's Hive CLI check can use an embedded metastore. Also query
the
+ # Thrift service used by Gravitino before allowing catalog tests to
start.
+ # A synchronous Docker exec and time.sleep cannot be bounded by
asyncio.wait_for.
+ command = [
+ "timeout",
+ "--signal=KILL",
+ f"{remaining}s",
+ "bash",
+ "-c",
+ "bash /tmp/check-status.sh && exec hive "
+ "--hiveconf hive.metastore.uris=thrift://localhost:9083 "
+ "-e 'show databases;'",
+ ]
try:
- command_and_args = ["bash", "/tmp/check-status.sh"]
- exec_result = hdfs_container.exec_run(command_and_args)
- if exec_result.exit_code != 0:
- message = (
- f"Command {command_and_args} exited with
{exec_result.exit_code}"
- )
- logger.warning(message)
- logger.warning("output: %s", exec_result.output)
- output_status_command = ["hdfs", "dfsadmin", "-report"]
- exec_result = hdfs_container.exec_run(output_status_command)
- logger.info("HDFS report, output: %s", exec_result.output)
- else:
- logger.info("HDFS startup successfully!")
- return True
- except DockerException as e:
- logger.error(
- "Exception occurred while checking HDFS container status: %s",
e
+ result = hdfs_container.exec_run(command)
+ last_output = result.output
+ if result.exit_code == 0:
+ logger.info("HDFS and Hive Metastore are ready")
+ return
+ logger.warning(
+ "HDFS/Hive readiness probe exited with %s: %s",
+ result.exit_code,
+ last_output,
)
- time.sleep(10)
- return False
-
-
-async def check_hdfs_container_status(hdfs_container):
- timeout_sec = 150
- try:
- result = await asyncio.wait_for(
- check_hdfs_status(hdfs_container), timeout=timeout_sec
- )
- if not result:
- raise InternalError("HDFS container startup failed!")
- except asyncio.TimeoutError as e:
- raise GravitinoRuntimeException(
- "Timeout occurred while waiting for checking HDFS container
status."
- ) from e
+ except DockerException as error:
+ last_output = str(error)
+ logger.warning("Failed to check HDFS/Hive readiness: %s", error)
+ remaining = deadline - time.monotonic()
+ if remaining > 0:
+ time.sleep(min(interval_sec, remaining))
+ raise GravitinoRuntimeException(
+ f"HDFS/Hive Metastore did not become ready within {timeout_sec}s. "
+ f"Last probe output: {last_output}"
+ )
class HDFSContainer(BaseContainer):
@@ -81,5 +83,9 @@ class HDFSContainer(BaseContainer):
super().__init__(container_name, image_name, environment)
- asyncio.run(check_hdfs_container_status(self._container))
- self._fetch_ip()
+ try:
+ check_hdfs_container_status(self._container)
+ self._fetch_ip()
+ except Exception:
+ self.close()
+ raise
diff --git a/clients/client-python/tests/unittests/test_hdfs_container.py
b/clients/client-python/tests/unittests/test_hdfs_container.py
new file mode 100644
index 0000000000..0400cadc18
--- /dev/null
+++ b/clients/client-python/tests/unittests/test_hdfs_container.py
@@ -0,0 +1,88 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+import unittest
+from unittest.mock import Mock, patch
+
+from docker.errors import DockerException
+from gravitino.exceptions.base import GravitinoRuntimeException
+from tests.integration.containers.hdfs_container import (
+ HDFSContainer,
+ check_hdfs_container_status,
+)
+
+MODULE = "tests.integration.containers.hdfs_container"
+
+
+class TestHDFSContainer(unittest.TestCase):
+ def test_waits_for_remote_metastore(self):
+ container = Mock()
+ container.exec_run.side_effect = [
+ Mock(exit_code=1, output=b"Metastore connection refused"),
+ Mock(exit_code=0, output=b"default"),
+ ]
+ with (
+ patch(f"{MODULE}.time.sleep") as sleep,
+ patch(f"{MODULE}.time.monotonic", return_value=0),
+ ):
+ check_hdfs_container_status(container)
+ self.assertEqual(2, container.exec_run.call_count)
+ sleep.assert_called_once()
+ command = container.exec_run.call_args.args[0]
+ self.assertEqual(["timeout", "--signal=KILL", "150s"], command[:3])
+ self.assertIn("hive.metastore.uris=thrift://localhost:9083",
command[-1])
+ self.assertIn("bash /tmp/check-status.sh &&", command[-1])
+ self.assertIn("show databases;", command[-1])
+
+ def test_deadline_bounds_probe_and_sleep(self):
+ container = Mock()
+ container.exec_run.return_value = Mock(exit_code=137, output=b"probe
timed out")
+ with (
+ patch(f"{MODULE}.time.monotonic", side_effect=[0, 0, 3, 5]),
+ patch(f"{MODULE}.time.sleep") as sleep,
+ ):
+ with self.assertRaisesRegex(GravitinoRuntimeException, "probe
timed out"):
+ check_hdfs_container_status(container, timeout_sec=5)
+ self.assertEqual("5s", container.exec_run.call_args.args[0][2])
+ sleep.assert_called_once_with(2)
+ container.exec_run.assert_called_once()
+
+ def test_retries_docker_errors(self):
+ container = Mock()
+ container.exec_run.side_effect = [
+ DockerException("temporary failure"),
+ Mock(exit_code=0, output=b"default"),
+ ]
+ with patch(f"{MODULE}.time.sleep"):
+ check_hdfs_container_status(container)
+ self.assertEqual(2, container.exec_run.call_count)
+
+ def test_removes_container_when_readiness_fails(self):
+ with (
+ patch.dict("os.environ", {"GRAVITINO_CI_HIVE_DOCKER_IMAGE":
"test-image"}),
+ patch(f"{MODULE}.BaseContainer.__init__", return_value=None),
+ patch(
+ f"{MODULE}.check_hdfs_container_status",
+ side_effect=RuntimeError("not ready"),
+ ),
+ patch.object(HDFSContainer, "close") as close,
+ patch.object(HDFSContainer, "_fetch_ip") as fetch_ip,
+ ):
+ with self.assertRaisesRegex(RuntimeError, "not ready"):
+ HDFSContainer()
+ close.assert_called_once()
+ fetch_ip.assert_not_called()
