Add wrapper for Metadata requests #497

[jewertow]

Purpose

Introduction of a wrapper for Metadata requests.

References

See #497

Changes

• Add MetadataClient

Background Context

The new feature to request Kafka Metadata from the internal KafkaConsumerActor (see #491) would be even nicer with thin wrappers (javadsl and scaladsl) hiding the ask calls and mapping to the correct response types.

[jewertow]

I think this feature will be useful for #886, so I would like to implement it first. At the moment I have prepared a draft of the target implementation. I need an opinion whether this is a good way or not. If this is a good approach, I will continue this way. Otherwise, please give me some hints.

[ennru]

Thank you for giving this some attention!

[ennru]

Yes, less magic is better.

[ennru]

This will throw an exception if there was an error and fail the future that way. Instead, you should transform the Future. To be Scala 2.11 compatible you need to do it like this

    .flatMap {
      case Success(res) => Future.successful(res)
      case Failure(e) => Future.failed(e)
    }(ExecutionContexts.sameThreadExecutionContext)

With this it is OK to use Akka's sameThreadExecutionContext to run the flatMap on.

[jewertow]

Do we really need this? I added failing test cases and the code behaves the same for:

.map(_.response.get)

as well as for:

.map(_.response)
.flatMap {
  case Success(res) => Future.successful(res)
  case Failure(exception) => Future.failed(exception)
}

[ennru]

To run it safely on the sameThreadExecutionContext we may not allow any exceptions happening in the operator.

[ennru]

Have you thought about making the MetadataClient an instance which would keep the actor ref and timeout? That would open up to have a convenience factory method which would take ConsumerSettings and create the actor internally (a bit like your first proposal).

[jewertow]

Yes, I think it's a good way.

[jewertow]

I pushed new proposal. Please look again and tell me if I can continue this way for the other requests.

[ennru]

It would be preferable to have one MetadataClient instance per consumer actor/settings.
Make the primary constructor private and let it take the consumer actor's ActorRef.
Create factory methods in the companion object, one taking the ActorRef and one taking ConsumerSettings.

[jewertow]

Thanks for the hint. I pushed new implementation.

[jewertow]

@ennru could you look at it again and say if you meant it?

[ennru]

Yes, that looks good.
The actor should only be stopped by the metadata client if it was created by it, you'd possibly need to pass that information to the client. So two constructors on that class might make sense.
close might be better than stop.

[ennru]

It would be great if you could get back to this to push it over the line.

[ennru]

The code looks good!
Please suggest an update to the docs in https://doc.akka.io/docs/alpakka-kafka/snapshot/consumer-metadata.html

[jewertow]

@ennru I pushed updated docs. Look at this please.

[ennru]

Looks great!
It would be better to show this documentation in consumer-metadata.md as this now becomes the recommened way to access the consumer metadata.

[jewertow]

I agree with you that it should be in the consumer-metadata.md documentation, but current implementation of MetadataClient does not support requests of the type GetOffsetsForTimes.
I'm going to add support for the GetOffsetsForTimes in the foreseeable future.
For now I would add to consumer-metadata docs the following statement:

@@@ warning

Since the `KafkaConsumerActor` is the internal mechanism, the recommended method to get metadata is using the [metadata-client](metadata-client.md).

[ennru]

Please join the new docs into the consumer-metadata.md page.
Show the MetadataClient first and have the raw experience after it.

[jewertow]

Done.

[ennru]

LGTM.

[ennru]

Thank you for implementing this!

[jewertow]

Thank you for the opportunity to contribute to alpakka-kafka. It was the best experience I’ve ever had in open source! Thank you for your help and advices.