code: public static long getPropAsLongFromSingleOrMultiWorkUnitState(WorkUnitState workUnitState,
                                                                   String key, int partitionId) {
      return Long.parseLong(workUnitState.contains(key) ? workUnitState.getProp(key)
          : workUnitState.getProp(KafkaUtils.getPartitionPropName(key, partitionId), "0"));
    }
docs: Get a property as long from a work unit that may or may not be a multiworkunit.
  This method is needed because the SingleLevelWorkUnitPacker does not squeeze work units
  into a multiworkunit, and thus does not append the partitionId to property keys, while
  the BiLevelWorkUnitPacker does.
  Return 0 as default if key not found in either form.