@Override
        protected void checkSequence(short producerEpoch, int appendFirstSeq, long offset) {
            if (currentEntry.isEmpty() && updatedEntry.isEmpty() && appendFirstSeq != 0
                // await sequence 0 message append timeout and retry
                && time.milliseconds() - createTimestamp < AWAIT_SEQ_ZERO_TIMEOUT
            ) {
                throw new OutOfOrderSequenceException("Invalid sequence number for new created log, producer " + producerId() + " " +
                    "at offset " + offset + " in partition " + topicPartition + ": " + producerEpoch + " (request epoch), " + appendFirstSeq + " (seq. number), " +
                    updatedEntry.producerEpoch() + " (current producer epoch)");
            }
            super.checkSequence(producerEpoch, appendFirstSeq, offset);
        }
Only keep the last snapshot which offset less than or equals to the recoveryPointCheckpoint
core/src/main/scala/kafka/log/streamaspect/ElasticProducerStateManager.java