code: private List<WorkUnit> getWorkUnitsForTopic(KafkaTopic topic, SourceState state, Optional<State> topicSpecificState) {
      Timer.Context context = this.metricContext.timer("isTopicQualifiedTimer").time();
      boolean topicQualified = isTopicQualified(topic);
      context.close();
  
      List<WorkUnit> workUnits = Lists.newArrayList();
      for (KafkaPartition partition : topic.getPartitions()) {
        WorkUnit workUnit = getWorkUnitForTopicPartition(partition, state, topicSpecificState);
        if (workUnit != null) {
          // For disqualified topics, for each of its workunits set the high watermark to be the same
          // as the low watermark, so that it will be skipped.
          if (!topicQualified) {
            skipWorkUnit(workUnit);
          }
          workUnits.add(workUnit);
        }
      }
      this.partitionsToBeProcessed.addAll(topic.getPartitions());
      return workUnits;
    }
docs: /*
  This function need to be thread safe since it is called in the Runnable