public void close(final Timer timer) {
        // we do not need to re-enable wakeups since we are closing already
        client.disableWakeups();
        try {
            maybeAutoCommitOffsetsSync(timer);
            while (pendingAsyncCommits.get() > 0 && timer.notExpired()) {
                ensureCoordinatorReady(timer);
                client.poll(timer);
                invokeCompletedOffsetCommitCallbacks();
            }
        } finally {
            super.close(timer);
        }
    }
@Test
    public void testLeaveGroupOnClose() {

        subscriptions.subscribe(singleton(topic1), Optional.of(rebalanceListener));
        joinAsFollowerAndReceiveAssignment(coordinator, singletonList(t1p));

        final AtomicBoolean received = new AtomicBoolean(false);
        client.prepareResponse(body -> {
            received.set(true);
            LeaveGroupRequest leaveRequest = (LeaveGroupRequest) body;
            return validateLeaveGroup(groupId, consumerId, leaveRequest);
        }, new LeaveGroupResponse(
            new LeaveGroupResponseData().setErrorCode(Errors.NONE.code())));
        coordinator.close(time.timer(0));
        assertTrue(received.get());
    }