Google-cloud-java: Pubsub : Publisher / Subscriber with service accounts

[hmigneron]

The com.google.cloud.pubsub namespace now being deprecated, I'm trying to use use the replacement (Publisher / Subscriber in com.google.cloud.pubsub.spi.v1).

I can't find an example of how to use service accounts though. With the deprecated version I would use

PubSubOptions
    .newBuilder()
    .setProjectId(Configurations.getGoogleCloudProjectId())
    .setCredentials(ServiceAccountCredentials.fromStream(new FileInputStream(Configurations.getGoogleCloudCredentials())))
    .build()
    .getService();

where Configurations.getGoogleCloudCredentials() points to a json file with the credentials.

What's the equivalent now ? How do we use service accounts (or really, any specific credentials besides the machine defaults) with Publisher / Subscriber ?

Thanks !

[pongad]

It's probably easiest if you have GOOGLE_APPLICATION_CREDENTIALS environment variable pointed to the json file. If that's not possible, this should work:

Publisher
  .defaultBuilder(topic)
  .setChannelProvider(TopicAdminSettings
    .defaultChannelProviderBuilder()
    .setCredentialsProvider(FixedCredentialsProvider.create(yourCredentialsHere))
    .build())
  .build();

I'll close the issue for now, but please reopen if the problem persists.

[abhijitdhar]

Hi @pongad @hmigneron do you know how to set the transport .setTransport(transport)
like we used to to with earlier API:
HttpTransport transport = new NetHttpTransport(); GoogleCredential credential = new GoogleCredential.Builder() .setTransport(transport) .setJsonFactory(JSON_FACTORY) .setServiceAccountId(serviceAccount) .setServiceAccountScopes(Arrays.asList("https://www.googleapis.com/auth/pubsub")) .setServiceAccountPrivateKeyFromP12File(new File(keyFile)) .build();

The call to publisher.publish(pubsubMessage).get() is hanging forever. I'm not sure what the problem is. I'm guessing its the transport setting

[pongad]

@abhijitdhar The new client uses gRPC, not HTTP, so HttpTransport won't work. In most use cases, InstantiatingChannelProvider should do what you want.

If you need something more advanced, you can use FixedChannelProvider to pass in a custom ManagedChannel

[abhijitdhar]

Thanks @pongad the authentication is working now. But publisher.publish(pubsubMessage).get() is hanging forever. Any idea what might be wrong? Posted the question here - https://stackoverflow.com/questions/44295892/com-google-cloud-pubsub-spi-v1-publisher-publish-is-not-sending-data-to-pubsub

[pongad]

@abhijitdhar Did you print the result of get() or anything like that? I tried running your code snippet on my machine. It looks like the problem is actually that get() itself returns, but your program doesn't terminate afterwards.

The reason is Publisher internally creates a thread pool to help it perform RPCs. After get() returns, the threads are still alive so the JVM doesn't exit. If you add publisher.shutdown() after get(), the program should exit properly.

Please let me know if this helps.

[abhijitdhar]

hi @pongad I think I may have found the problem, but don't know how it works with gRPC.
I have to set the proxy and the way I was doing it earlier was:
System.setProperty("https.proxyHost", proxyHost); System.setProperty("https.proxyPort", proxyPort); System.setProperty("https.nonProxyHosts", nonProxyHosts);
I am guessing that doesn't work with new API that does not use HTTP.
Any idea how to do that here?

[pongad]

This went beyond my expertise. @carl-mastrangelo could you help point us in the right direction?

[carl-mastrangelo]

@abhijitdhar Proxy support is in gRPC, but I don't think it is fully featured yet. Can you try setting the GRPC_PROXY_EXP environment variable to set the proxy?

[abhijitdhar]

hi @carl-mastrangelo Thanks a lot! GRPC_PROXY_EXP worked great!

[RedShift1]

@pongad Code sample doesn't work. "defaultChannelProviderBuilder" method on TopicAdminSettings doesn't exist (using version 0.21.1-alpha).

[pongad]

Ah! It looks like the gax-http refactoring broke us. The following code should work now though.

Publisher
  .defaultBuilder(topic)
  .setCredentialsProvider(FixedCredentialsProvider.create(yourCredentialsHere))
  .build();

We should probably put this in an example Java file somewhere so it actually gets compile checked.
@garrettjonesgoogle I have my hands a little full on Go fixit week. Could someone on Java fixit pick this up? If not, I can loop back to this a little later.

[garrettjonesgoogle]

We shouldn't keep reopening old issues whenever there are breaking changes with the code in them. I have opened a new issue https://github.com/GoogleCloudPlatform/google-cloud-java/issues/2322 to add a sample for this use case. Closing the present issue.

[harisatyanarayana-php]

hi I am not able to read the messages from subscription , if I pass my credentials in the following below way :

Credentials credentials = null;
try {
credentials = ServiceAccountCredentials.fromStream(new FileInputStream("MyJsonFile"));

    } catch (IOException e) {
        System.out.println( e.getMessage());
        //logger.error("Could not import Google Pubsub json credentials: {}" + "\n" + e.getMessage());
    }
    CredentialsProvider credentialsProvider = FixedCredentialsProvider.create(credentials);


Subscriber subscriber = null;
subscriber = Subscriber.newBuilder(subscriptionName, receiver).setCredentialsProvider(credentialsProvider)
        .build();
  // Start the subscriber.
  subscriber.startAsync().awaitRunning();
  System.out.printf("Listening for messages on %s:\n", subscriptionName.toString());
  // Allow the subscriber to run for 30s unless an unrecoverable error occurs.
  subscriber.awaitTerminated();

}
output error :

Exception in thread "main" java.lang.IllegalStateException: Expected the service InnerService [FAILED] to be RUNNING, but the service has FAILED
at com.google.common.util.concurrent.AbstractService.checkCurrentState(AbstractService.java:379)
at com.google.common.util.concurrent.AbstractService.awaitRunning(AbstractService.java:303)
at com.google.api.core.AbstractApiService.awaitRunning(AbstractApiService.java:96)
at com.example.pubsub.pusub_emulator.SubscribeAsyncExample.subscribeAsyncExample(SubscribeAsyncExample.java:70)
at com.example.pubsub.pusub_emulator.SubscribeAsyncExample.main(SubscribeAsyncExample.java:28)
Caused by: java.lang.IllegalAccessError: tried to access field io.opencensus.trace.unsafe.ContextUtils.CONTEXT_SPAN_KEY from class io.grpc.internal.CensusTracingModule$TracingClientInterceptor
at io.grpc.internal.CensusTracingModule$TracingClientInterceptor.interceptCall(CensusTracingModule.java:384)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at io.grpc.internal.CensusStatsModule$StatsClientInterceptor.interceptCall(CensusStatsModule.java:690)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.api.gax.grpc.GrpcChannelUUIDInterceptor.interceptCall(GrpcChannelUUIDInterceptor.java:52)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.api.gax.grpc.GrpcHeaderInterceptor.interceptCall(GrpcHeaderInterceptor.java:80)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at com.google.api.gax.grpc.GrpcMetadataHandlerInterceptor.interceptCall(GrpcMetadataHandlerInterceptor.java:55)
at io.grpc.ClientInterceptors$InterceptorChannel.newCall(ClientInterceptors.java:156)
at io.grpc.internal.ManagedChannelImpl.newCall(ManagedChannelImpl.java:789)
at io.grpc.internal.ForwardingManagedChannel.newCall(ForwardingManagedChannel.java:63)
at com.google.api.gax.grpc.GrpcClientCalls.newCall(GrpcClientCalls.java:90)
at com.google.api.gax.grpc.GrpcDirectBidiStreamingCallable.internalCall(GrpcDirectBidiStreamingCallable.java:62)
at com.google.api.gax.grpc.GrpcExceptionBidiStreamingCallable.internalCall(GrpcExceptionBidiStreamingCallable.java:63)
at com.google.api.gax.tracing.TracedBidiCallable.internalCall(TracedBidiCallable.java:87)
at com.google.api.gax.rpc.BidiStreamingCallable$4.internalCall(BidiStreamingCallable.java:256)
at com.google.api.gax.rpc.BidiStreamingCallable.splitCall(BidiStreamingCallable.java:187)
at com.google.cloud.pubsub.v1.StreamingSubscriberConnection.initialize(StreamingSubscriberConnection.java:190)
at com.google.cloud.pubsub.v1.StreamingSubscriberConnection.doStart(StreamingSubscriberConnection.java:112)
at com.google.api.core.AbstractApiService$InnerService.doStart(AbstractApiService.java:148)
at com.google.common.util.concurrent.AbstractService.startAsync(AbstractService.java:249)
at com.google.api.core.AbstractApiService.startAsync(AbstractApiService.java:120)
at com.google.cloud.pubsub.v1.Subscriber.startConnections(Subscriber.java:361)
at com.google.cloud.pubsub.v1.Subscriber.startStreamingConnections(Subscriber.java:332)
at com.google.cloud.pubsub.v1.Subscriber.access$1300(Subscriber.java:92)
at com.google.cloud.pubsub.v1.Subscriber$2.run(Subscriber.java:279)
at java.lang.Thread.run(Thread.java:748)