KAFKA-20426: Using both group.id and assign() causes a busy loop in AsyncKafkaConsumer

[brandboat]

When a consumer uses manual assignment (assign()) instead of group
subscription, the member remains in the UNSUBSCRIBED state. In this
state, heartbeats are skipped.

Previously, because the heartbeat interval was initialized to 0, the
maximumTimeToWait calculation would return 0 when heartbeats were
skipped. This caused pollForFetches to return immediately and enter a
busy-loop, consuming excessive CPU.

This patch fixes the issue by ensuring maximumTimeToWait returns
Long.MAX_VALUE whenever shouldSkipHeartbeat() is true.

[JiayaoS]

Thanks! LGTM.

[JiayaoS]

I was wondering if we should handle the pollTimer.isExpired() case here, as shouldSkipHeartbeat() being true would cause the logic to return Long.MAX_VALUE even if the timer has expired. However, I noticed that the callers use Math.min() with timer.remainingMs(), so it won't impact production. Ive also run the unit tests locally and everything looks solid.

[JiayaoS]

I wrote a local test case for this.

@Test
    public void testAssignBusyLoop() {
        time = Time.SYSTEM;
        consumer = newConsumer();
        TopicPartition tp = new TopicPartition("foo", 3);
        AtomicInteger loopCount = new AtomicInteger(0);
        doAnswer(invocation -> {
            loopCount.incrementAndGet();
            return Fetch.empty();
        }).when(fetchCollector).collectFetch(any(FetchBuffer.class));
        doReturn(LeaderAndEpoch.noLeaderOrEpoch()).when(metadata).currentLeader(any());
        completeAssignmentChangeEventSuccessfully();
        consumer.assign(singleton(tp));
        completeAsyncPollEventSuccessfully();
        consumer.poll(Duration.ofSeconds(1));
        assertTrue(loopCount.get() < 300,
                "Busy loop detected! loopCount:" + loopCount.get());
    }

Unit Test Output: Busy loop detected! loopCount:78668

Proves that it indeed causes a busy loop.

Subsequently, I used
doAnswer(invocation -> Long.MAX_VALUE).when(applicationEventHandler).maximumTimeToWait();
to mock of maximumTimeToWait() return Long.MAX_VALUE, and it worked perfectly.