code: 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;
    }
docs: Helper method to check on stream start that some kind of state is initialized to avoid a common error
  of starting without initializing.