synchronized void promoteDelayedOffsetSyncs() {
        pendingOffsetSyncs.putAll(delayedOffsetSyncs);
        delayedOffsetSyncs.clear();
    }
@Test
    public void testPromoteDelayedOffsetSyncs() {
        int maxOffsetLag = 50;
        @SuppressWarnings("unchecked")
        KafkaProducer<byte[], byte[]> producer = mock(KafkaProducer.class);
        Semaphore outstandingOffsetSyncs = new Semaphore(1);

        OffsetSyncWriter offsetSyncWriter = new OffsetSyncWriter(producer, topicName, outstandingOffsetSyncs, maxOffsetLag);
        offsetSyncWriter.maybeQueueOffsetSyncs(topicPartition, 0, 100);
        offsetSyncWriter.maybeQueueOffsetSyncs(topicPartition, 1, 101);
        offsetSyncWriter.promoteDelayedOffsetSyncs();

        assertTrue(offsetSyncWriter.getDelayedOffsetSyncs().isEmpty());
        Map<TopicPartition, OffsetSync> pendingOffsetSyncs = offsetSyncWriter.getPendingOffsetSyncs();
        assertEquals(1, pendingOffsetSyncs.size());
        assertEquals(1, pendingOffsetSyncs.get(topicPartition).upstreamOffset());
        assertEquals(101, pendingOffsetSyncs.get(topicPartition).downstreamOffset());
    }