env.setControlEventHandler(new ControlEventHandler() {
@Override
public void onEvent(ByteBuf event) {
if (DcpSnapshotMarkerMessage.is(event)) {
short partition = DcpSnapshotMarkerMessage.partition(event);
<BUG>PartitionState ps = sessionState.get(partition);
</BUG>
ps.setSnapshotStartSeqno(DcpSnapshotMarkerMessage.startSeqno(event));
ps.setSnapshotEndSeqno(DcpSnapshotMarkerMessage.endSeqno(event));
<BUG>sessionState.set(partition, ps);
</BUG>
acknowledgeBuffer(event);

env.setControlEventHandler(new ControlEventHandler() {
@Override
public void onEvent(ByteBuf event) {
if (DcpSnapshotMarkerMessage.is(event)) {
short partition = DcpSnapshotMarkerMessage.partition(event);
PartitionState ps = sessionState().get(partition);
ps.setSnapshotStartSeqno(DcpSnapshotMarkerMessage.startSeqno(event));
ps.setSnapshotEndSeqno(DcpSnapshotMarkerMessage.endSeqno(event));
sessionState().set(partition, ps);
acknowledgeBuffer(event);
