private List < Short > selectInitializedPartitions ( int clusterPartitions , List < Short > partitions ) { List < Short > initializedPartitions = new ArrayList <> ( ) ; SessionState state = sessionState ( ) ; for ( short partition : partitions ) { PartitionState ps = state . get ( partition ) ; if ( ps != null ) { if ( MathUtils . lessThanUnsigned ( ps . getStartSeqno ( ) , ps . getEndSeqno ( ) ) ) { initializedPartitions . add ( partition ) ; } else { LOGGER . debug ( "Skipping partition {}, because startSeqno({}) >= endSeqno({})" , partition , ps . getStartSeqno ( ) , ps . getEndSeqno ( ) ) ; } } else { LOGGER . debug ( "Skipping partition {}, because its state is null" , partition ) ; } } if ( initializedPartitions . size ( ) > clusterPartitions ) { throw new IllegalStateException ( "Session State has " + initializedPartitions + " partitions while the cluster has " + clusterPartitions + "!" ) ; } return initializedPartitions ; }
Helper method to check on stream start that some kind of state is initialized to avoid a common error of starting without initializing .