[SPARK-32468][SS][TESTS] Fix timeout config issue in Kafka connector tests

[gaborgsomogyi]

What changes were proposed in this pull request?

While I'm implementing SPARK-32032 I've found a bug in Kafka: https://issues.apache.org/jira/browse/KAFKA-10318. This will cause issues only later when it's fixed but it would be good to fix it now because SPARK-32032 would like to bring in AdminClient where the code blows up with the mentioned ConfigException. This would reduce the code changes in the mentioned jira. In this PR I've changed default.api.timeout.ms to request.timeout.ms which fulfils this condition.

Why are the changes needed?

Solve later problems and reduce SPARK-32032 PR size.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Existing unit tests.

[gaborgsomogyi]

cc @HeartSaVioR @zsxwing @xuanyuanking

[SparkQA]

Test build #126722 has finished for PR 29272 at commit 4c3a9f1.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[dongjoon-hyun]

According to KAFKA-10318, is this Kafka 2.5.0 only bug?

[gaborgsomogyi]

No, just only the latest added. It effects all previous versions where these params exist.

[HeartSaVioR]

So my understanding is that request.timeout.ms should take precedence over default.api.timeout.ms, and it's desired for end users (in apps) to configure request.timeout.ms instead of default.api.timeout.ms, though they need to increase default.api.timeout.ms as well if they'd like to set request.timeout.ms higher than default.api.timeout.ms. Do I understand correctly?

[gaborgsomogyi]

@HeartSaVioR that's exactly the main message. This part must be enforced on Kafka side just like in case of AdminClient:
they need to increase default.api.timeout.ms as well if they'd like to set request.timeout.ms higher than default.api.timeout.ms

[HeartSaVioR]

LGTM

[HeartSaVioR]

Looks like there's no further input. I'll retrigger test and merge once it passes.

[HeartSaVioR]

retest this, please

[HeartSaVioR]

retest this, please

[HeartSaVioR]

retest this, please

[SparkQA]

Test build #126814 has finished for PR 29272 at commit 4c3a9f1.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[HeartSaVioR]

retest this, please

[SparkQA]

Test build #126827 has finished for PR 29272 at commit 4c3a9f1.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[HeartSaVioR]

retest this, please

[HyukjinKwon]

retest this please

[HyukjinKwon]

@gaborgsomogyi can you sync with the master and rebase? then the GitHub Actions build should pass. Jenkins seems down for some reasons.

[SparkQA]

Test build #126842 has finished for PR 29272 at commit 4c3a9f1.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #126840 has finished for PR 29272 at commit 4c3a9f1.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[HeartSaVioR]

lol, somehow it passes Jenkins build. Merging.

[HeartSaVioR]

Thanks, merged to master!

[gaborgsomogyi]

Thanks everybody to help me!

[HeartSaVioR]

I have been getting reports for flaky tests based on this - I took a closer look on the explanation of Kafka configurations, and realized they're not interchangeable. My bad.

This test (in Kafka) perfectly describes how these configurations work together (I'm wondering why default.api.timeout.ms is not used there and it just passes the value into timeout parameter, but nothing changed to show the behavior):

https://github.com/apache/kafka/blob/28b7d8e21656649fb09b09f9bacfe865b0ca133c/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java#L4086-L4127

    @Test
    public void testSuccessfulRetryAfterRequestTimeout() throws Exception {
        HashMap<Integer, Node> nodes = new HashMap<>();
        MockTime time = new MockTime();
        Node node0 = new Node(0, "localhost", 8121);
        nodes.put(0, node0);
        Cluster cluster = new Cluster("mockClusterId", nodes.values(),
                Arrays.asList(new PartitionInfo("foo", 0, node0, new Node[]{node0}, new Node[]{node0})),
                Collections.emptySet(), Collections.emptySet(),
                Collections.emptySet(), nodes.get(0));

        final int requestTimeoutMs = 1000;
        final int retryBackoffMs = 100;
        final int apiTimeoutMs = 3000;

        try (AdminClientUnitTestEnv env = new AdminClientUnitTestEnv(time, cluster,
                AdminClientConfig.RETRY_BACKOFF_MS_CONFIG, String.valueOf(retryBackoffMs),
                AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, String.valueOf(requestTimeoutMs))) {
            env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());

            final ListTopicsResult result = env.adminClient()
                    .listTopics(new ListTopicsOptions().timeoutMs(apiTimeoutMs));

            // Wait until the first attempt has been sent, then advance the time
            TestUtils.waitForCondition(() -> env.kafkaClient().hasInFlightRequests(),
                    "Timed out waiting for Metadata request to be sent");
            time.sleep(requestTimeoutMs + 1);

            // Wait for the request to be timed out before backing off
            TestUtils.waitForCondition(() -> !env.kafkaClient().hasInFlightRequests(),
                    "Timed out waiting for inFlightRequests to be timed out");
            time.sleep(retryBackoffMs);

            // Since api timeout bound is not hit, AdminClient should retry
            TestUtils.waitForCondition(() -> env.kafkaClient().hasInFlightRequests(),
                    "Failed to retry Metadata request");
            env.kafkaClient().respond(prepareMetadataResponse(cluster, Errors.NONE));

            assertEquals(1, result.listings().get().size());
            assertEquals("foo", result.listings().get().iterator().next().name());
        }
    }

When we request without timeout parameter, we're now using default value of default.api.timeout.ms which is 60 seconds, longer than test timeout. If the request can't be retried, then request.timeout.ms would make the request fail in same time (which is intended). But if the request can be retried, it's no longer the same and it may hold the request until it is timed out via default.api.timeout.ms (which is not expected).

So if the change is intended to make sure default.api.timeout.ms >= request.timeout.ms (see KAFKA-10318), both configurations should be changed together.