static MessageId findFirstLedgerWithinThreshold(List<PersistentTopicInternalStats.LedgerInfo> ledgers,
                                                    long sizeThreshold) {
        long suffixSize = 0L;

        ledgers = Lists.reverse(ledgers);
        long previousLedger = ledgers.get(0).ledgerId;
        for (PersistentTopicInternalStats.LedgerInfo l : ledgers) {
            suffixSize += l.size;
            if (suffixSize > sizeThreshold) {
                return new MessageIdImpl(previousLedger, 0L, -1);
            }
            previousLedger = l.ledgerId;
        }
        return null;
    }
@Test
    public void testFindFirstLedgerWithinThreshold() throws Exception {
        List<LedgerInfo> ledgers = new ArrayList<>();
        ledgers.add(newLedger(0, 10, 1000));
        ledgers.add(newLedger(1, 10, 2000));
        ledgers.add(newLedger(2, 10, 3000));

        // test huge threshold
        Assert.assertNull(CmdTopics.findFirstLedgerWithinThreshold(ledgers, Long.MAX_VALUE));

        // test small threshold
        Assert.assertEquals(CmdTopics.findFirstLedgerWithinThreshold(ledgers, 0),
                            new MessageIdImpl(2, 0, -1));

        // test middling thresholds
        Assert.assertEquals(CmdTopics.findFirstLedgerWithinThreshold(ledgers, 1000),
                            new MessageIdImpl(2, 0, -1));
        Assert.assertEquals(CmdTopics.findFirstLedgerWithinThreshold(ledgers, 5000),
                            new MessageIdImpl(1, 0, -1));
    }