Wrap MetricsControl access to consumer in future

[ennru]

Fixes #528

[2m]

LGTM. Great that the access to the ActorRef was hidden behind a method. So changing it to Future[ActorRef] is not going to change use API.

[2m]

Needs a rebase.

[s4nk]

@ennru I'm calling metrics after materializing a stream and I'm getting an exception like this:

java.lang.IllegalStateException: not yet initialized: only setHandler is allowed in GraphStageLogic constructor. To access materializer use Source/Flow/Sink.setup factory
	at akka.stream.stage.GraphStageLogic.interpreter(GraphStage.scala:393)
	at akka.stream.stage.GraphStageLogic.materializer(GraphStage.scala:403)
	at akka.kafka.internal.BaseSingleSourceLogic.executionContext(BaseSingleSourceLogic.scala:34)
	at akka.kafka.internal.MetricsControl.metrics(ControlImplementations.scala:88)
	at akka.kafka.internal.MetricsControl.metrics$(ControlImplementations.scala:78)
	at akka.kafka.internal.BaseSingleSourceLogic.metrics(BaseSingleSourceLogic.scala:26)

The flow takes messages from a KafkaConsumerActor and sends them to another actor, the whole thing looks like this:

      val sink: Sink[Output, NotUsed] =
        ActorSink.actorRefWithAck[Output, SinkActorCommandType, Ack](...)

      val control: Consumer.Control =
        Consumer.plainExternalSource[String, Array[Byte]](kafkaConsumerActor, subscription)
          .map { consumerRecord =>
            processRecord(consumerRecord)
          }
          .to(sink)
          .run()

      control.metrics

Did the race condition just move to a different place now?

[s4nk]

OK, looks like somebody already raised the issue, #847