void firePendingOffsetSyncs() {
        while (true) {
            OffsetSync pendingOffsetSync;
            synchronized (this) {
                Iterator<OffsetSync> syncIterator = pendingOffsetSyncs.values().iterator();
                if (!syncIterator.hasNext()) {
                    // Nothing to sync
                    LOG.trace("No more pending offset syncs");
                    return;
                }
                pendingOffsetSync = syncIterator.next();
                if (!outstandingOffsetSyncs.tryAcquire()) {
                    // Too many outstanding syncs
                    LOG.trace("Too many in-flight offset syncs; will try to send remaining offset syncs later");
                    return;
                }
                syncIterator.remove();
            }
            // Publish offset sync outside of synchronized block; we may have to
            // wait for producer metadata to update before Producer::send returns
            sendOffsetSync(pendingOffsetSync);
            LOG.trace("Dispatched offset sync for {}", pendingOffsetSync.topicPartition());
        }
    }
@Test
    public void testFirePendingOffsetSyncs() {
        int maxOffsetLag = 1;

        Semaphore outstandingOffsetSyncs = new Semaphore(1);

        OffsetSyncWriter offsetSyncWriter = new OffsetSyncWriter(producer, topicName, outstandingOffsetSyncs, maxOffsetLag);

        offsetSyncWriter.maybeQueueOffsetSyncs(topicPartition, 0, 100);
        assertEquals(offsetSyncWriter.partitionStates().get(topicPartition).lastSyncDownstreamOffset, 100);

        offsetSyncWriter.firePendingOffsetSyncs();

        ArgumentCaptor<Callback> producerCallback = ArgumentCaptor.forClass(Callback.class);
        when(producer.send(any(), producerCallback.capture())).thenAnswer(mockInvocation -> {
            producerCallback.getValue().onCompletion(null, null);
            return null;
        });

        // We should have dispatched this sync to the producer
        verify(producer, times(1)).send(any(), any());

        offsetSyncWriter.maybeQueueOffsetSyncs(topicPartition, 2, 102);
        assertEquals(offsetSyncWriter.partitionStates().get(topicPartition).lastSyncDownstreamOffset, 102);
        offsetSyncWriter.firePendingOffsetSyncs();

        // in-flight offset syncs; will not try to send remaining offset syncs immediately
        verifyNoMoreInteractions(producer);
    }