ci(ingest/kafka): improve kafka integration test reliability (datahub…

…-project#8085)

metadata-ingestion/setup.cfg

@@ -83,6 +83,11 @@ markers =
 testpaths =
     tests/unit
     tests/integration
+filterwarnings =
+    # Ignore some warnings that come from dependencies.
+    ignore:Deprecated call to \`pkg_resources.declare_namespace:DeprecationWarning
+    ignore:pkg_resources is deprecated as an API:DeprecationWarning
+    ignore:Did not recognize type:sqlalchemy.exc.SAWarning
 
 [coverage:run]
 # Because of some quirks in the way setup.cfg, coverage.py, pytest-cov,

metadata-ingestion/tests/integration/kafka/docker-compose.yml

@@ -5,6 +5,7 @@ services:
     image: confluentinc/cp-zookeeper:7.2.0
     env_file: zookeeper.env
     hostname: test_zookeeper
+    container_name: test_zookeeper
     ports:
       - "52181"
     volumes:

metadata-ingestion/tests/integration/kafka/test_kafka.py

@@ -15,6 +15,12 @@
 def test_kafka_ingest(docker_compose_runner, pytestconfig, tmp_path, mock_time):
     test_resources_dir = pytestconfig.rootpath / "tests/integration/kafka"
 
+    with docker_compose_runner(
+        test_resources_dir / "docker-compose.yml", "kafka", cleanup=False
+    ) as docker_services:
+        wait_for_port(docker_services, "test_zookeeper", 52181, timeout=120)
+
+    # Running docker compose twice, since the broker sometimes fails to come up on the first try.
     with docker_compose_runner(
         test_resources_dir / "docker-compose.yml", "kafka"
     ) as docker_services:

metadata-ingestion/tests/test_helpers/docker_helpers.py

@@ -1,10 +1,13 @@
 import contextlib
+import logging
 import subprocess
 from typing import Callable, Optional, Union
 
 import pytest
 import pytest_docker.plugin
 
+logger = logging.getLogger(__name__)
+
 
 def is_responsive(container_name: str, port: int, hostname: Optional[str]) -> bool:
     """A cheap way to figure out if a port is responsive on a container"""
@@ -37,6 +40,7 @@ def wait_for_port(
             if checker
             else lambda: is_responsive(container_name, container_port, hostname),
         )
+        logger.info(f"Container {container_name} is ready!")
     finally:
         # use check=True to raise an error if command gave bad exit code
         subprocess.run(f"docker logs {container_name}", shell=True, check=True)

metadata-ingestion/tests/unit/test_packaging.py

@@ -1,6 +1,11 @@
+import pytest
+
 import datahub as datahub_metadata
 
 
+@pytest.mark.filterwarnings(
+    "ignore:pkg_resources is deprecated as an API:DeprecationWarning"
+)
 def test_datahub_version():
     # Simply importing pkg_resources checks for unsatisfied dependencies.
     import pkg_resources