Support Pub/Sub Emulator

[fkorotkov]

It seems right now google-cloud-pubsub doesn't support gcloud emulator. At least I'm not able to configure it for Pub/Sub even though I managed to use an emulator with google-cloud-datastore. I think it is related to googleapis/google-cloud-datastore#114.

Anyways, when I'm trying to create a subscriber using a channel provider with no credentials. Here are some snippets in Kotlin of how I create a SubscriptionAdminClient:

val channelProvider = InstantiatingChannelProvider.newBuilder()
        .setEndpoint("localhost:$port")
        .setCredentialsProvider(FixedCredentialsProvider.create(NoCredentials.getInstance()))
        .build()

protected fun getSubscriptionAdminClient(): SubscriptionAdminClient {
  val adminSettings = SubscriptionAdminSettings.defaultBuilder()
    .setChannelProvider(channelProvider)
    .build()
  return SubscriptionAdminClient.create(adminSettings)
}

When I'm trying to call com.google.cloud.pubsub.spi.v1.SubscriptionAdminClient#createSubscription I'm getting:

com.google.api.gax.grpc.ApiException: io.grpc.StatusRuntimeException: UNKNOWN
	at com.google.api.gax.grpc.ExceptionTransformingCallable$ExceptionTransformingFuture.onFailure(ExceptionTransformingCallable.java:109)
	at com.google.api.gax.core.ApiFutures$1.onFailure(ApiFutures.java:53)
	at com.google.common.util.concurrent.Futures$4.run(Futures.java:1126)
	at com.google.common.util.concurrent.MoreExecutors$DirectExecutor.execute(MoreExecutors.java:399)
	at com.google.common.util.concurrent.AbstractFuture.executeListener(AbstractFuture.java:902)
	at com.google.common.util.concurrent.AbstractFuture.addListener(AbstractFuture.java:636)
	at com.google.common.util.concurrent.ForwardingListenableFuture.addListener(ForwardingListenableFuture.java:45)
	at com.google.api.gax.core.internal.ApiFutureToListenableFuture.addListener(ApiFutureToListenableFuture.java:53)
	at com.google.common.util.concurrent.Futures.addCallback(Futures.java:1138)
	at com.google.common.util.concurrent.Futures.addCallback(Futures.java:1076)
	at com.google.api.gax.core.ApiFutures.addCallback(ApiFutures.java:48)
	at com.google.api.gax.grpc.ExceptionTransformingCallable.futureCall(ExceptionTransformingCallable.java:65)
	at com.google.api.gax.grpc.RetryingCallable$GrpcRetryCallable.call(RetryingCallable.java:139)
	at com.google.api.gax.grpc.RetryingCallable.futureCall(RetryingCallable.java:84)
	at com.google.api.gax.grpc.UnaryCallable.futureCall(UnaryCallable.java:219)
	at com.google.api.gax.grpc.UnaryCallable.futureCall(UnaryCallable.java:230)
	at com.google.api.gax.grpc.UnaryCallable.call(UnaryCallable.java:258)
	at com.google.cloud.pubsub.spi.v1.SubscriptionAdminClient.createSubscription(SubscriptionAdminClient.java:339)
	at com.google.cloud.pubsub.spi.v1.SubscriptionAdminClient.createSubscription(SubscriptionAdminClient.java:307)
	at com.fkorotkov.gcloud.pubsub.PubSubImpl.registerSubscriber(PubSubImpl.kt:72)
Caused by: io.grpc.StatusRuntimeException: UNKNOWN
	at io.grpc.Status.asRuntimeException(Status.java:540)
	at io.grpc.stub.ClientCalls$UnaryStreamToFuture.onClose(ClientCalls.java:439)
	at io.grpc.ClientInterceptors$CheckedForwardingClientCall.start(ClientInterceptors.java:202)
	at io.grpc.ForwardingClientCall.start(ForwardingClientCall.java:47)
	at com.google.api.gax.grpc.HeaderInterceptor$1.start(HeaderInterceptor.java:62)
	at io.grpc.stub.ClientCalls.startCall(ClientCalls.java:270)
	at io.grpc.stub.ClientCalls.asyncUnaryRequestCall(ClientCalls.java:249)
	at io.grpc.stub.ClientCalls.futureUnaryCall(ClientCalls.java:186)
	at com.google.api.gax.grpc.DirectCallable.futureCall(DirectCallable.java:59)
	at com.google.api.gax.grpc.ExceptionTransformingCallable.futureCall(ExceptionTransformingCallable.java:62)
	... 55 more
Caused by: java.lang.IllegalStateException: OAuth2Credentials instance does not support refreshing the access token. An instance with a new access token should be used, or a derived type that supports refreshing.
	at com.google.auth.oauth2.OAuth2Credentials.refreshAccessToken(OAuth2Credentials.java:182)
	at com.google.auth.oauth2.OAuth2Credentials.refresh(OAuth2Credentials.java:149)
	at com.google.auth.oauth2.OAuth2Credentials.getRequestMetadata(OAuth2Credentials.java:135)
	at io.grpc.auth.ClientAuthInterceptor.getRequestMetadata(ClientAuthInterceptor.java:148)
	at io.grpc.auth.ClientAuthInterceptor.access$100(ClientAuthInterceptor.java:62)
	at io.grpc.auth.ClientAuthInterceptor$1.checkedStart(ClientAuthInterceptor.java:94)
	at io.grpc.ClientInterceptors$CheckedForwardingClientCall.start(ClientInterceptors.java:194)
	... 62 more

As I mentioned above I'm able to use Datastore emulator when I run it with --no-legacy flag and by creating a Datastore service like this:

DatastoreOptions.newBuilder()
      .setProjectId("unit-testing")
      .setHost("localhost:$port")
      .setCredentials(NoCredentials.getInstance())
      .build()
      .getService()

I'm using the latest available version of the libraries at the moment:

{
  "group": "com.google.cloud",
  "name": "google-cloud-datastore",
  "version": "0.13.0-beta"
},
{
  "group": "com.google.cloud",
  "name": "google-cloud-pubsub",
  "version": "0.13.0-alpha"
}

[pongad]

I can replicate this on my desktop; it looks like a bug.

This seems to work correctly however:

ManagedChannel chan = ManagedChannelBuilder.forAddress("localhost", 8085).usePlaintext(true).build();

SubscriptionAdminSettings settings =
    SubscriptionAdminSettings.defaultBuilder()
        .setChannelProvider(FixedChannelProvider.create(chan))
        .build();

try (SubscriptionAdminClient subscriptionAdminClient =
    SubscriptionAdminClient.create(settings)) {
  subscriptionAdminClient.createSubscription(
      subscription, topic, PushConfig.getDefaultInstance(), 0);
}

@tcoffee-google Since you're working on auth-related stuff, could you take a look?

[fkorotkov]

@pongad @tcoffee-google thank you for taking a look!

I've created a reproducible example if you'll need to: fkorotkov/snippets@a0dd999

Also interesting that if I change pubsub version to 0.17.1-alpha the test case simply hangs. I can see that a container is running through docker ps and by getting logs via docekr logs CONTAINER_ID I can see bunch of:

[pubsub] May 01, 2017 4:38:42 PM io.gapi.emulators.netty.HttpVersionRoutingHandler channelRead
[pubsub] INFO: Detected non-HTTP/2 connection.
[pubsub] May 01, 2017 4:38:42 PM io.gapi.emulators.netty.NotFoundHandler handleRequest
[pubsub] INFO: Unknown request URI: /bad-request
[pubsub] May 01, 2017 4:38:42 PM io.gapi.emulators.grpc.GrpcServer$3 operationComplete
[pubsub] INFO: Adding handler(s) to newly registered Channel.
[pubsub] May 01, 2017 4:38:42 PM io.gapi.emulators.netty.HttpVersionRoutingHandler channelRead
[pubsub] INFO: Detected non-HTTP/2 connection.

Will play around with 0.17.1-alpha a bit more. Great speed of releases!

[pongad]

@fkorotkov I can also reproduce that. I suspect that's actually gRPC retrying a connection; NoCredentials is probably causing an infinite retry behavior somehow.

The fix I added above seems to solve both issues though. Is it good enough to unblock you for now?

[fkorotkov]

@pongad yeah, works perfectly for me fkorotkov/snippets@d310bf9. Thank you very much! Going to incorporate it into the main app(using an in-memory in-house implementation at the moment).

[pongad]

This can be fixed by adding setPlainText to InstantiatingChannelProvider but it would have to call ManagedChannelBuilder::usePlainText which is beta API.

@garrettjonesgoogle ?

[garrettjonesgoogle]

I would prefer not to add the dependency on the beta API within our non-test code (InstantiatingChannelProvider).

[rcleveng]

I tried the workarounds in code and was able to send a message, however never received it using a pull subscription.