public boolean commitOffsetsSync(Map<TopicPartition, OffsetAndMetadata> offsets, Timer timer) {
        invokeCompletedOffsetCommitCallbacks();

        if (offsets.isEmpty()) {
            // We guarantee that the callbacks for all commitAsync() will be invoked when
            // commitSync() completes, even if the user tries to commit empty offsets.
            return invokePendingAsyncCommits(timer);
        }

        long attempts = 0L;
        do {
            if (coordinatorUnknownAndUnreadySync(timer)) {
                return false;
            }

            RequestFuture<Void> future = sendOffsetCommitRequest(offsets);
            client.poll(future, timer);

            // We may have had in-flight offset commits when the synchronous commit began. If so, ensure that
            // the corresponding callbacks are invoked prior to returning in order to preserve the order that
            // the offset commits were applied.
            invokeCompletedOffsetCommitCallbacks();

            if (future.succeeded()) {
                if (interceptors != null)
                    interceptors.onCommit(offsets);
                return true;
            }

            if (future.failed() && !future.isRetriable())
                throw future.exception();

            timer.sleep(retryBackoff.backoff(attempts++));
        } while (timer.notExpired());

        return false;
    }
@Test
    public void testPrepareJoinAndRejoinAfterFailedRebalance() {
        final List<TopicPartition> partitions = singletonList(t1p);
        try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, false, Optional.of("group-id"), true)) {
            coordinator.ensureActiveGroup();

            prepareOffsetCommitRequest(singletonMap(t1p, 100L), Errors.REBALANCE_IN_PROGRESS);

            assertThrows(RebalanceInProgressException.class, () -> coordinator.commitOffsetsSync(
                singletonMap(t1p, new OffsetAndMetadata(100L)),
                time.timer(Long.MAX_VALUE)));

            assertFalse(client.hasPendingResponses());
            assertFalse(client.hasInFlightRequests());

            int generationId = 42;
            String memberId = "consumer-42";

            client.prepareResponse(joinGroupFollowerResponse(generationId, memberId, "leader", Errors.NONE));

            MockTime time = new MockTime(1);

            // onJoinPrepare will be executed and onJoinComplete will not.
            boolean res = coordinator.joinGroupIfNeeded(time.timer(100));

            assertFalse(res);
            assertFalse(client.hasPendingResponses());
            // SynGroupRequest not responded.
            assertEquals(1, client.inFlightRequestCount());
            assertEquals(generationId, coordinator.generation().generationId);
            assertEquals(memberId, coordinator.generation().memberId);

            // Imitating heartbeat thread that clears generation data.
            coordinator.maybeLeaveGroup("Clear generation data.");

            assertEquals(AbstractCoordinator.Generation.NO_GENERATION, coordinator.generation());

            client.respond(syncGroupResponse(partitions, Errors.NONE));

            // Join future should succeed but generation already cleared so result of join is false.
            res = coordinator.joinGroupIfNeeded(time.timer(1));

            assertFalse(res);

            // should have retried sending a join group request already
            assertFalse(client.hasPendingResponses());
            assertEquals(1, client.inFlightRequestCount());

            // Retry join should then succeed
            client.respond(joinGroupFollowerResponse(generationId, memberId, "leader", Errors.NONE));
            client.prepareResponse(syncGroupResponse(partitions, Errors.NONE));

            res = coordinator.joinGroupIfNeeded(time.timer(3000));

            assertTrue(res);
            assertFalse(client.hasPendingResponses());
            assertFalse(client.hasInFlightRequests());
        }
        Collection<TopicPartition> lost = getLost(partitions);
        assertEquals(lost.isEmpty() ? null : lost, rebalanceListener.lost);
        assertEquals(lost.size(), rebalanceListener.lostCount);
    }