public void forceDrain() {
        appendLock.lock();
        try {
            drainStatus = DrainStatus.STARTED;
            maybeCompleteDrain();
        } finally {
            appendLock.unlock();
        }
    }
@Test
    public void testForceDrain() {
        int leaderEpoch = 17;
        long baseOffset = 157;
        int lingerMs = 50;
        int maxBatchSize = 512;

        Mockito.when(memoryPool.tryAllocate(maxBatchSize))
            .thenReturn(ByteBuffer.allocate(maxBatchSize));

        BatchAccumulator<String> acc = buildAccumulator(
            leaderEpoch,
            baseOffset,
            lingerMs,
            maxBatchSize
        );

        List<String> records = asList("a", "b", "c", "d", "e", "f", "g", "h", "i");

        // Append records
        assertEquals(baseOffset, acc.append(leaderEpoch, records.subList(0, 1), false));
        assertEquals(baseOffset + 2, acc.append(leaderEpoch, records.subList(1, 3), false));
        assertEquals(baseOffset + 5, acc.append(leaderEpoch, records.subList(3, 6), false));
        assertEquals(baseOffset + 7, acc.append(leaderEpoch, records.subList(6, 8), false));
        assertEquals(baseOffset + 8, acc.append(leaderEpoch, records.subList(8, 9), false));

        assertFalse(acc.needsDrain(time.milliseconds()));
        acc.forceDrain();
        assertTrue(acc.needsDrain(time.milliseconds()));
        assertEquals(0, acc.timeUntilDrain(time.milliseconds()));

        // Drain completed batches
        List<BatchAccumulator.CompletedBatch<String>> batches = acc.drain();

        assertEquals(1, batches.size());
        assertFalse(acc.needsDrain(time.milliseconds()));
        assertEquals(Long.MAX_VALUE - time.milliseconds(), acc.timeUntilDrain(time.milliseconds()));

        BatchAccumulator.CompletedBatch<String> batch = batches.get(0);
        assertEquals(records, batch.records.get());
        assertEquals(baseOffset, batch.baseOffset);
        assertEquals(time.milliseconds(), batch.appendTimestamp());
    }