return Observable
.from(partitions)
.flatMap(new Func1<Integer, Observable<?>>() {
@Override
public Observable<?> call(Integer partition) {
<BUG>PartitionState partitionState = sessionState.get(partition);
</BUG>
return conductor.startStreamForPartition(
partition.shortValue(),
partitionState.getUuid(),

return Observable
.from(partitions)
.flatMap(new Func1<Integer, Observable<?>>() {
@Override
public Observable<?> call(Integer partition) {
PartitionState partitionState = sessionState().get(partition);
return conductor.startStreamForPartition(
partition.shortValue(),
partitionState.getUuid(),
