env.setDataEventHandler(new DataEventHandler() {
@Override
public void onEvent(ByteBuf event) {
if (DcpMutationMessage.is(event)) {
short partition = DcpMutationMessage.partition(event);
<BUG>PartitionState ps = sessionState.get(partition);
</BUG>
ps.setStartSeqno(DcpMutationMessage.revisionSeqno(event));
<BUG>sessionState.set(partition, ps);
</BUG>
} else if (DcpDeletionMessage.is(event)) {

env.setDataEventHandler(new DataEventHandler() {
@Override
public void onEvent(ByteBuf event) {
if (DcpMutationMessage.is(event)) {
short partition = DcpMutationMessage.partition(event);
PartitionState ps = sessionState().get(partition);
ps.setStartSeqno(DcpMutationMessage.revisionSeqno(event));
sessionState().set(partition, ps);
} else if (DcpDeletionMessage.is(event)) {
