refactor(test): Use Kafka consumer offsets API for lag

[anshbansal]

No description provided.

[alwaysmeticulous]

✅ Meticulous spotted 0 visual differences across 992 screens tested: view results.

Meticulous evaluated ~8 hours of user flows against your PR.

_{Expected differences? Click here. Last updated for commit d14644d. This comment will update as new commits are pushed.}

[codecov]

❌ 1 Tests Failed:

Tests completed	Failed	Passed	Skipped
6025	1	6024	42

View the top 1 failed test(s) by shortest run time

tests.managed_ingestion.managed_ingestion_test::test_create_list_get_ingestion_execution_request

Stack Traces | 754s run time

auth_session = <tests.utils.TestSessionWrapper object at 0x7fa956a79410>

    @pytest.mark.dependency(
        depends=[
            "test_create_list_get_remove_ingestion_source",
        ]
    )
    def test_create_list_get_ingestion_execution_request(auth_session):
        # Create new ingestion source
        query = """mutation createIngestionSource($input: UpdateIngestionSourceInput!) {\n
                createIngestionSource(input: $input)
            }"""
        variables: Dict[str, Any] = {
            "input": {
                "name": "My Test Ingestion Source",
                "type": "mysql",
                "description": "My ingestion source description",
                "schedule": {"interval": "*/5 * * * *", "timezone": "UTC"},
                "config": {
                    "recipe": '{"source":{"type":"mysql","config":{"include_tables":true,"database":null,"password":"${MYSQL_PASSWORD}","profiling":{"enabled":false},"host_port":null,"include_views":true,"username":"${MYSQL_USERNAME}"}},"pipeline_name":"urn:li:dataHubIngestionSource:f38bd060-4ea8-459c-8f24-a773286a2927"}',
                    "version": "0.8.18",
                    "executorId": "mytestexecutor",
                },
            }
        }
        res_data = execute_graphql(auth_session, query, variables)
        assert res_data["data"]["createIngestionSource"] is not None
    
        ingestion_source_urn = res_data["data"]["createIngestionSource"]
    
        # Create a request to execute the ingestion source
        query = """mutation createIngestionExecutionRequest($input: CreateIngestionExecutionRequestInput!) {\n
                createIngestionExecutionRequest(input: $input)
            }"""
        variables = {"input": {"ingestionSourceUrn": ingestion_source_urn}}
        res_data = execute_graphql(auth_session, query, variables)
        assert res_data["data"]["createIngestionExecutionRequest"] is not None, (
            f"res_data was {res_data}"
        )
    
        execution_request_urn = res_data["data"]["createIngestionExecutionRequest"]
    
        res_data = _ensure_ingestion_source_present(auth_session, ingestion_source_urn, 1)
    
        ingestion_source = res_data["data"]["ingestionSource"]
    
>       assert (
            ingestion_source["executions"]["executionRequests"][0]["urn"]
            == execution_request_urn
        )
E       AssertionError: assert 'urn:li:dataHubExecutionRequest:239c0b82-4a87-4dbc-8413-59fb1a7b9e31' == 'urn:li:dataHubExecutionRequest:a511e909-d5f0-4b09-b1e7-e7a2c8b1031e'
E         
E         - urn:li:dataHubExecutionRequest:a511e909-d5f0-4b09-b1e7-e7a2c8b1031e
E         + urn:li:dataHubExecutionRequest:239c0b82-4a87-4dbc-8413-59fb1a7b9e31

tests/managed_ingestion/managed_ingestion_test.py:565: AssertionError

To view more test analytics, go to the Test Analytics Dashboard
_{📋 Got 3 mins? Take this short survey to help us improve Test Analytics.}

[david-leifker]

1 ms interval and the underlying method is skipCache by default. This is not ideal. The get_kafka_consumer_offsets() method should not be disabling cache by default. The main loop is too tight, increased busy work, and potentially misses something if the lag goes to zero momentarily.