title: Flink get RuntimeContext
body: <p>In my Flink code I am using a custom input format, which throws an exception. It seems I need an instance of <code>RuntimeContext</code>, but how can I get one?</p>
  
  <p>My format class looks like this:</p>
  
  <pre class="lang-scala prettyprint-override"><code>MyInputFormat extends org.apache.flink.api.common.io.DelimitedInputFormat[T]{
  @transient var lineCounter: IntCounter = _
  override def open(split: FileInputSplit): Unit = {
      super.open(split)
      lineCounter = new IntCounter()
      getRuntimeContext.addAccumulator("rowsInFile", lineCounter) // this line throws IllegalStateException
  </code></pre>
  
  <p>My main program looks like this:</p>
  
  <pre class="lang-scala prettyprint-override"><code>val env = ExecutionEnvironment.getExecutionEnvironment
  val format = new MyInputFormat
  env.readFile(format, inputFile.getAbsolutePath) // throws exception
  </code></pre>
  
  <p>The exception that gets thrown:</p>
  
  <pre class="lang-scala prettyprint-override"><code>java.lang.IllegalStateException: The runtime context has not been initialized yet. Try accessing it in one of the other life cycle methods.
      at org.apache.flink.api.common.io.RichInputFormat.getRuntimeContext(RichInputFormat.java:51)
  </code></pre>
  
  <p>My class needs a <code>RuntimeContext</code> because it extends <code>DelimitedInputFormat</code> which extends... <code>RichInputFormat</code></p>
  
  <pre class="lang-scala prettyprint-override"><code>public abstract class DelimitedInputFormat&lt;OT&gt; extends FileInputFormat&lt;OT&gt;
  public abstract class FileInputFormat&lt;OT&gt; extends RichInputFormat&lt;OT, FileInputSplit&gt;
  public abstract class RichInputFormat&lt;OT, T extends InputSplit&gt; implements InputFormat&lt;OT, T&gt;
      private transient RuntimeContext runtimeContext;
      public void setRuntimeContext(RuntimeContext t)
      public RuntimeContext getRuntimeContext()
  </code></pre>
  
  <p>So any instance of <code>RichInputFormat</code> expects us to <code>setRuntimeContext(RuntimeContext t)</code> after it's created.</p>
  
  <p>I expect I should be doing the following:</p>
  
  <pre class="lang-scala prettyprint-override"><code>val env = ExecutionEnvironment.getExecutionEnvironment
  val runtimeContext: RuntimeContext = ??? // How do I get this?
  val format = new MyInputFormat
  format.setRuntimeContext(runtimeContext)
  env.readFile(format, inputFile.getAbsolutePath) // no longer throws exception
  </code></pre>
  
  <p>But how do I get an instance of RuntimeContext?
  The exception gets thrown because my custom input format does not have a <code>RuntimeContext</code>. I would set one, but I don't know where to get it.</p>