.doOnNext(new Action1<long[]>() {
@Override
public void call(long[] longs) {
short partition = (short) longs[0];
long seqno = longs[1];
<BUG>PartitionState partitionState = sessionState.get(partition);
</BUG>
partitionState.setStartSeqno(seqno);
partitionState.setSnapshotStartSeqno(seqno);
<BUG>sessionState.set(partition, partitionState);
</BUG>
}

.doOnNext(new Action1<long[]>() {
@Override
public void call(long[] longs) {
short partition = (short) longs[0];
long seqno = longs[1];
PartitionState partitionState = sessionState().get(partition);
partitionState.setStartSeqno(seqno);
partitionState.setSnapshotStartSeqno(seqno);
sessionState().set(partition, partitionState);
}
