public void removeExpiredProducers(long currentTimeMs) {
        producers.entrySet().removeIf(entry -> isProducerExpired(currentTimeMs, entry.getValue()));
        producerIdCount = producers.size();

        verificationStates.entrySet().removeIf(entry ->
            (currentTimeMs - entry.getValue().timestamp()) >= producerStateManagerConfig.producerIdExpirationMs()
        );
    }
@Test
    public void testProducerStateAfterFencingAbortMarker() {
        appendClientEntry(stateManager, producerId, epoch, defaultSequence, 0L, true);
        appendEndTxnMarker(stateManager, producerId, (short) (epoch + 1), ControlRecordType.ABORT, 1L);

        ProducerStateEntry lastEntry = getLastEntryOrElseThrownByProducerId(stateManager, producerId);
        assertEquals(OptionalLong.empty(), lastEntry.currentTxnFirstOffset());
        assertEquals(-1, lastEntry.lastDataOffset());
        assertEquals(-1, lastEntry.firstDataOffset());

        // The producer should not be expired because we want to preserve fencing epochs
        stateManager.removeExpiredProducers(time.milliseconds());
        assertDoesNotThrow(() -> getLastEntryOrElseThrownByProducerId(stateManager, producerId));
    }