title: Continuously stream HadoopInput files off of s3/gcs into Flink
body: <p>I have a custom <code>HadoopInput</code> type that I want to read off of S3 into Flink as new records hit the bucket.</p>
  <p>Here's how it currently works in batch.</p>
  <pre><code>    val path = &quot;s3://path/&quot;
      val hadoopInput = new CustomInputType().asInstanceOf[InputFormat[LongWritable,Text]]
  
      val job = Job.getInstance()
      FileInputFormat.addInputPath(job, new Path(path))
  
      val customInput  = HadoopInputs.createHadoopInput[LongWritable,Text](hadoopInput, classOf[LongWritable], classOf[Text], job)
      val parameters = new Configuration()
      parameters.setBoolean(&quot;recursive.file.enumeration&quot;,true)
      customInput.configure(parameters)
  
      
      
      val lines = streamExecutionEnvironment.createInput(customInput)
  </code></pre>
  <p>I want to be able to replicate the code:</p>
  <pre><code>val stream: DataStream[String] = streamExecutionEnvironment.readFile(textInputFormat, \ 
                                          path, FileProcessingMode.PROCESS_CONTINUOUSLY, 1000)
  </code></pre>
  <p>But with the HadoopInput instead of the TextInput.</p>
  <p>What methods are available to me?</p>