protect against null counters

The `getCounter` method of the `Reporter` returned from `HadoopFlowProcess` was
returning null in some cases for a few jobs that we run in production. (It is
unclear why these jobs were seeing null counters.)

From looking at the Hadoop source code, getCounter does return null in some instances,
in particular the Reporter.NULL implementation unconditionally returns null from
its getCounter implementation. Hadoop does this despite not documenting that null
is a valid return value.

Solution: Null check the return value of `Reporter.getCounter` to workaround the issue.

Fixes #1716

scalding-core/src/main/scala/com/twitter/scalding/Stats.scala

@@ -4,6 +4,7 @@ import cascading.flow.{ Flow, FlowListener, FlowDef, FlowProcess }
 import cascading.flow.hadoop.HadoopFlowProcess
 import cascading.stats.CascadingStats
 import java.util.concurrent.ConcurrentHashMap
+import org.apache.hadoop.mapreduce.Counter
 import org.slf4j.{ Logger, LoggerFactory }
 import scala.collection.JavaConverters._
 import scala.collection.mutable
@@ -64,8 +65,12 @@ private[scalding] final case class GenericFlowPCounterImpl(fp: FlowProcess[_], s
 }
 
 private[scalding] final case class HadoopFlowPCounterImpl(fp: HadoopFlowProcess, statKey: StatKey) extends CounterImpl {
-  private[this] val cntr = fp.getReporter().getCounter(statKey.group, statKey.counter)
-  override def increment(amount: Long): Unit = cntr.increment(amount)
+  private[this] val counter: Option[Counter] = (for {
+    r <- Option(fp.getReporter)
+    c <- Option(r.getCounter(statKey.group, statKey.counter))
+  } yield c)
+
+  override def increment(amount: Long): Unit = counter.foreach(_.increment(amount))
 }
 
 object Stat {