void maybeQueueOffsetSyncs(TopicPartition topicPartition, long upstreamOffset, long downstreamOffset) {
        PartitionState partitionState =
                partitionStates.computeIfAbsent(topicPartition, x -> new PartitionState(maxOffsetLag));

        OffsetSync offsetSync = new OffsetSync(topicPartition, upstreamOffset, downstreamOffset);
        if (partitionState.update(upstreamOffset, downstreamOffset)) {
            // Queue this sync for an immediate send, as downstream state is sufficiently stale
            synchronized (this) {
                delayedOffsetSyncs.remove(topicPartition);
                pendingOffsetSyncs.put(topicPartition, offsetSync);
            }
            partitionState.reset();
        } else {
            // Queue this sync to be delayed until the next periodic offset commit
            synchronized (this) {
                delayedOffsetSyncs.put(topicPartition, offsetSync);
            }
        }
    }
@Test
    public void testMaybeQueueOffsetSyncs() {
        int maxOffsetLag = 2;

        @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, 1);
        assertFalse(offsetSyncWriter.getDelayedOffsetSyncs().containsKey(topicPartition));
        assertTrue(offsetSyncWriter.getPendingOffsetSyncs().containsKey(topicPartition));
        assertEquals(offsetSyncWriter.partitionStates().get(topicPartition).lastSyncDownstreamOffset, 1);

        offsetSyncWriter.maybeQueueOffsetSyncs(topicPartition, 1, 2);
        assertTrue(offsetSyncWriter.getDelayedOffsetSyncs().containsKey(topicPartition));
        assertEquals(offsetSyncWriter.partitionStates().get(topicPartition).lastSyncDownstreamOffset, 1);
    }