title: Luigi dependencies specification issue with a separate task
body: <p>I have <code>3</code> <code>Luigi</code> tasks: first generates an output file that is written to <code>hadoop</code>, second - uses this output file to load it into <code>Elasticsearch</code>, third one - gets a completely separate file and also loads it into <code>Elasticsearch</code>. Third task is rather disconnected from the first two, but I want it to be run when the first two are finished. There can be multiple files (of the same type) fed into the first task, so in the second one I specify the dependencies like that:</p>
  
  <pre><code>def requires(self):
      return [SeqrVCFToMTTask()]
  </code></pre>
  
  <p><code>SeqrVCFToMTTask</code> being the first task. It works fine, so first two tasks run perfectly. Now, when I try to specify dependency in the same way in the third task:</p>
  
  <pre><code>def requires(self):
      return [SeqrMTToESTask()]
  
  (`SeqrMTToESTask` - name of the second task).
  </code></pre>
  
  <p>It fails with the error:</p>
  
  <blockquote>
    <p>raise HTTP_EXCEPTIONS.get(status_code, TransportError)(status_code, error_message, additional_info)
    elasticsearch.exceptions.RequestError: TransportError(400, '')</p>
  </blockquote>
  
  <p>and the final <code>luigi</code> task output looks like that:</p>
  
  <pre><code>===== Luigi Execution Summary =====
  
  Scheduled 3 tasks of which:
  * 1 complete ones were encountered:
      - 1 SeqrVCFToMTTask(...)
  * 1 failed:
      - 1 SeqrMTToESTask(...)
  * 1 were left pending, among these:
      * 1 had failed dependencies:
          - 1 SeqrGenesQCToESTask(...)
  
  This progress looks :( because there were failed tasks
  
  ===== Luigi Execution Summary =====
  </code></pre>
  
  <p>I know it has nothing to do with <code>Elasticsearch</code> because the <code>3rd</code> task - <code>SeqrGenesQCToESTask</code> - runs fine separately when the dependencies are omitted (just uncommenting <code>requires</code>). How should I specify dependencies here correctly? The only thing that I need is for the <code>3rd</code> task to start running after the first two are finished. </p>
  
  <blockquote>
    <p>Update</p>
  </blockquote>
  
  <p>More detailed code:</p>
  
  <pre><code>class SeqrVCFToMTTask(HailMatrixTableTask):
      reference_ht_path = luigi.Parameter(...
      ...
      ...
  
      def run(self):
          self.read_vcf_write_mt()
  
      def read_vcf_write_mt(self, schema_cls=SeqrVariantsAndGenotypesSchema):
      ...
      ...
  
  class SeqrMTToESTask(HailElasticSearchTask):
      dest_file = luigi.Parameter()
      def __init__(self, *args, **kwargs):
          # TODO: instead of hardcoded index, generate from project_guid, etc.
          super().__init__(*args, **kwargs)
  
      def requires(self):
          return [SeqrVCFToMTTask()]
  
      def output(self):
          filename = self.dest_file
          return getTarget(filename)
  
      def run(self):
          mt = self.import_mt()
          row_table = SeqrVariantsAndGenotypesSchema.elasticsearch_row(mt)
          self.export_table_to_elasticsearch(row_table, self._mt_num_shards(mt))
  
  class SeqrGenesQCToESTask(luigi.Task):
      source_path = luigi.Parameter(...
      ...
      ...
  
      def __init__(self, *args, **kwargs):
          super().__init__(*args, **kwargs)
          self._es = ElasticsearchClient(host=self.es_host, port=self.es_port)
  
      def requires(self):
          return [SeqrMTToESTask()]
  
      def output(self):
          # TODO: Use https://luigi.readthedocs.io/en/stable/api/luigi.contrib.esindex.html.
          filename = self.dest_file
          return getTarget(filename)
  
      def run(self):
          // Doing some data transformations, then 
          // Exporting data to Elasticsearch
  
  
  class HailMatrixTableTask(luigi.Task):
       source_paths = luigi.Parameter(...
       ...
       ...
  
      def __init__(self, *args, **kwargs):
          super().__init__(*args, **kwargs)
  
          try:
              # Locally it should be '_FrozenOrderedDict' but on AWS for an unknown reason only 'FrozenOrderedDict' works
              self.source_paths = list(json.loads(self.source_paths, object_pairs_hook=luigi.parameter._FrozenOrderedDict))
          except json.JSONDecodeError:
              self.source_paths = [self.source_paths]
  
      def requires(self):
          # We only exclude globs in source path here so luigi does not check if the file exists
          return [VcfFile(filename=s) for s in self.source_paths if '*' not in s]
  
      def output(self):
          filename = self.dest_path
          return getTarget(filename)
  
      def complete(self):
          # Complete is called by Luigi to check if the task is done and will skip if it is.
          # By default it checks to see that the output exists, but we want to check for the
          # _SUCCESS file to make sure it was not terminated halfway.
          filename = self.dest_path
          full_path = os.path.join(filename, '_SUCCESS')
          return getTarget(full_path).exists()
  
      def run(self):
          // Import file, then output in different format
  
  class HailElasticSearchTask(luigi.Task):
      project_guid = luigi.Parameter(...
      ...
      ...
  
      def __init__(self, *args, **kwargs):
          super().__init__(*args, **kwargs)
          if not self.es_index:
              self.es_index = compute_index_name(args=public_class_props(self))
          self._es = ElasticsearchClient(host=self.es_host, port=self.es_port)
  
      def requires(self):
          return [VcfFile(filename=self.source_path)]
  
      def run(self):
          mt = self.import_mt()
          # TODO: Load into ES
  
      def import_mt(self):
          return hl.read_matrix_table(self.input()[0].path)
  
      def export_table_to_elasticsearch(self, table, num_shards):
          // Exports to ES index
  </code></pre>
  
  <blockquote>
    <p>Update</p>
  </blockquote>
  
  <p>I suspect it happens because the <code>3rd</code> task - <code>SeqrGenesQCToESTask</code> - opens up connection to <code>Elasticsearch</code> and then the <code>2nd</code> task - <code>SeqrMTToESTask</code> - can't output to <code>Elasticsearch</code> due to that since the error happens there, in <code>SeqrMTToESTask</code> on the line:</p>
  
  <pre><code>self.export_table_to_elasticsearch(row_table, self._mt_num_shards(mt))
  </code></pre>