KAFKA-18310: Flaky AbstractCoordinatorTest

[FrankYang0529]

The AbstractCoordinator#joinGroupIfNeeded calls ConsumerNetworkClient#poll(RequestFuture, Timer) [0]. It calls another method with disableWakeup = false. If joinFuture is not done, ConsumerNetworkClient does poll multiple times before timer is expired [1]. If joinFuture finishes too fast, the SyncGroupRequest can't be handled in AbstractCoordinator#ensureActiveGroup, so it doesn't throw WakeupException. Change ConsumerNetworkClient#poll(RequestFuture, Timer, boolean) like following in the trunk branch and the flaky cases can be reproduced frequently.

public boolean poll(RequestFuture<?> future, Timer timer, boolean disableWakeup) {
    do {
        poll(timer, future, disableWakeup);
        try {
            System.err.println("sleeping.....");
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    } while (!future.isDone() && timer.notExpired());
    return future.isDone();
}

[0]

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractCoordinator.java

Lines 480 to 481 in 3d49159

	final RequestFuture<ByteBuffer> future = initiateJoinGroup();
	client.poll(future, timer);

[1]

kafka/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java

Lines 230 to 235 in 3d49159

	public boolean poll(RequestFuture<?> future, Timer timer, boolean disableWakeup) {
	do {
	poll(timer, future, disableWakeup);
	} while (!future.isDone() && timer.notExpired());
	return future.isDone();
	}

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)