// Create QueryApi
val queryApi = DataClient().queryApi(catalogHrn)

// Define Kafka consumer settings assigning an id for the consumer (if no id is specified, a random UUID is assigned)
val consumerSettings =
  ConsumerSettings(groupName = "test-consumer", consumerId = "consumer-id")

// Subscribe to a layer
val subscription: Future[Subscription] = queryApi.subscribe("stream-layer", consumerSettings)

// Retrieve Kafka metrics
val kafkaMetrics: Future[Map[String, String]] =
  subscription.flatMap(s => s.subscriptionControl.getKafkaMetrics())

// Create QueryApi
QueryApi queryApi = DataClient.get(myActorSystem).queryApi(catalogHrn);

// Define Kafka consumer settings assigning an id for the consumer (if no id is specified, a
// random UUID is assigned)
ConsumerSettings consumerSettings =
    new ConsumerSettings.Builder()
        .withGroupName("test-consumer")
        .withConsumerId("consumer-id")
        .build();

// Subscribe to a layer
CompletionStage<Subscription> subscriptionFuture =
    queryApi.subscribe("stream-layer", consumerSettings);

// Retrieve Kafka metrics
subscriptionFuture.thenApply(
    subscription -> {
      scala.collection.immutable.List<String>
          stringList = // import scala.collection.JavaConverters;
          JavaConverters.asScalaBufferConverter(new ArrayList<String>()).asScala().toList();
      return subscription.getSubscriptionControl().getKafkaMetrics(stringList);
    });