private void appendEndTxnMarker(ProducerStateManager stateManager,
                                    long producerId,
                                    short producerEpoch,
                                    ControlRecordType controlType,
                                    long offset,
                                    int coordinatorEpoch,
                                    long timestamp) {
        ProducerAppendInfo producerAppendInfo = stateManager.prepareUpdate(producerId, AppendOrigin.COORDINATOR);
        EndTransactionMarker endTxnMarker = new EndTransactionMarker(controlType, coordinatorEpoch);
        Optional<CompletedTxn> completedTxn = producerAppendInfo.appendEndTxnMarker(endTxnMarker, producerEpoch, offset, timestamp);
        stateManager.update(producerAppendInfo);
        completedTxn.ifPresent(stateManager::completeTxn);
        stateManager.updateMapEndOffset(offset + 1);
    }
Test that marking a snapshot for deletion when the file has already been deleted
returns None instead of the SnapshotFile. The snapshot file should be removed from
the in-memory state of the ProducerStateManager. This scenario can occur during log
recovery when the intermediate ProducerStateManager instance deletes a file without
updating the state of the "real" ProducerStateManager instance which is passed to the Log.
storage/src/test/java/org/apache/kafka/storage/internals/log/ProducerStateManagerTest.java