code: public Flow getFlowByJobID(String cluster, String jobId,
        boolean populateTasks) throws IOException {
      Flow flow = null;
      JobKey key = idService.getJobKeyById(new QualifiedJobId(cluster, jobId));
      if (key != null) {
        byte[] startRow =
            ByteUtil.join(Constants.SEP_BYTES, Bytes.toBytes(key.getCluster()),
                Bytes.toBytes(key.getUserName()), Bytes.toBytes(key.getAppId()),
                Bytes.toBytes(key.getEncodedRunId()), Constants.EMPTY_BYTES);
  
        LOG.info("Reading job_history rows start at "
            + Bytes.toStringBinary(startRow));
        Scan scan = new Scan();
        // start scanning history at cluster!user!app!run!
        scan.setStartRow(startRow);
        // require that all results match this flow prefix
        scan.setFilter(new WhileMatchFilter(new PrefixFilter(startRow)));
  
        List<Flow> flows = createFromResults(scan, populateTasks, 1);
        if (flows.size() > 0) {
          flow = flows.get(0);
        }
      }
      return flow;
    }
docs: Returns the {@link Flow} instance containing the given job ID.
  
  @param cluster the cluster identifier
  @param jobId the job identifier
  @return