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 shouldLoseAllOwnedPartitionsBeforeRejoiningAfterResettingGenerationId() {
        final List<TopicPartition> partitions = singletonList(t1p);
        try (ConsumerCoordinator coordinator = prepareCoordinatorForCloseTest(true, false, Optional.of("group-id"), true)) {
            final Time realTime = Time.SYSTEM;
            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)));

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

            client.prepareResponse(joinGroupFollowerResponse(generationId, memberId, "leader", Errors.NONE));
            client.prepareResponse(syncGroupResponse(Collections.emptyList(), Errors.ILLEGAL_GENERATION));

            boolean res = coordinator.joinGroupIfNeeded(realTime.timer(1000));

            assertFalse(res);
            assertEquals(AbstractCoordinator.Generation.NO_GENERATION.generationId, coordinator.generation().generationId);
            assertEquals(AbstractCoordinator.Generation.NO_GENERATION.protocolName, coordinator.generation().protocolName);
            // member ID should not be reset
            assertEquals(memberId, coordinator.generation().memberId);

            res = coordinator.joinGroupIfNeeded(realTime.timer(1000));
            assertFalse(res);
        }
        Collection<TopicPartition> lost = getLost(partitions);
        assertEquals(lost.isEmpty() ? 0 : 1, rebalanceListener.lostCount);
        assertEquals(lost.isEmpty() ? null : lost, rebalanceListener.lost);
    }