code: private WorkUnit createEmptyWorkUnit(KafkaPartition partition, long previousOffset, long previousFetchEpochTime,
        Optional<State> topicSpecificState) {
      Offsets offsets = new Offsets();
      offsets.setEarliestOffset(previousOffset);
      offsets.setLatestOffset(previousOffset);
      offsets.startAtEarliestOffset();
      offsets.setOffsetFetchEpochTime(previousFetchEpochTime);
      return getWorkUnitForTopicPartition(partition, offsets, topicSpecificState);
    }
docs: thread safe