fix that get next offset while search messages.
when asked for an invalid offset, ask last elements to reset.
distribute partitions to parallelisms on storm