Not able to send message to pub/sub emulator via java/springboot app

[beena-yatin-kanyal]

I am using the pub/sub emulator to emulate the pub-sub feature of the google-cloud platform before going live in our spring boot application.

Step[1] - running the emulator -
gcloud beta emulators pubsub start --project="$PUBSUB_PROJECT_ID" --host-port="$PUBSUB_EMULATOR_HOST"

-- output --

Executing: cmd /c C:\Program Files (x86)\Google\Cloud SDK\google-cloud-sdk\platform\pubsub-emulator\bin\cloud-pubsub-emulator.bat --host=localhost --port=5189
[pubsub] This is the Google Pub/Sub fake.
[pubsub] Implementation may be incomplete or differ from the real system.
[pubsub] Oct 29, 2021 12:58:51 PM com.google.cloud.pubsub.testing.v1.Main main
[pubsub] INFO: IAM integration is disabled. IAM policy methods and ACL checks are not supported
[pubsub] SLF4J: Failed to load class "org.slf4j.impl.StaticLoggerBinder".
[pubsub] SLF4J: Defaulting to no-operation (NOP) logger implementation
[pubsub] SLF4J: See http://www.slf4j.org/codes.html#StaticLoggerBinder for further details.
[pubsub] Oct 29, 2021 12:58:53 PM com.google.cloud.pubsub.testing.v1.Main main
[pubsub] INFO: Server started, listening on 5189

Step[2] - creating the topics/subscription on the emulator using the python repo

python publisher.py my-test-project-id create my-topic-1
python subscriber.py my-test-project-id create-push my-topic-1 my-subscription-1 http://localhost:9700/oranges

-- output --

Created topic: projects/my-test-project-id/topics/my-topic-1
Push subscription created: name: "projects/my-test-project-id/subscriptions/my-subscription-1"
topic: "projects/my-test-project-id/topics/my-topic-1"
push_config {
  push_endpoint: "http://localhost:9700/oranges"
}
ack_deadline_seconds: 10
message_retention_duration {
  seconds: 604800
}
.
Endpoint for subscription is: http://localhost:9700/oranges

Step[3] - sending the message via the springboot app.

@Override
public String sendMessageToPubSubTopic(Request request) {
ByteString byteString = JsonUtil.convert(request);
try {
  PubsubMessage pubsubMessage = PubsubMessage.newBuilder()
	  .setData(byteString)
	  .build();
  log.info("Sending pub/sub message to topic= {} in project= {}",
	  publisher.getTopicName().getTopic(), publisher.getTopicName().getProject());
  return publisher.publish(pubsubMessage).get();
} catch (Exception exception) {
  log.warn("error");
  Thread.currentThread().interrupt();
  throw new Exception("some-bad-exception-occured");
}
}

Now once I send the message to the emulator I get the below exception as soon as the get() method is called in the return statement. can someone advise what is wrong here or with the emulator?

{"ts":"10 29 2021 14:24:43.602","th":"http-nio-9700-exec-1","request":"","":"some_name","api":"POST api/my","user":"dc202734-3895-11ec-b53c-f09e4ac3134cser","level":"INFO","logger":"com.office.cloud.service.impl.PubSubServiceImpl","msg":"Sending pub/sub message to topic= emulator--management-topic in project= my-test-project-id"}
{"ts":"10 29 2021 14:24:47.533","th":"http-nio-9700-exec-1","request":"","":"some_name","api":"POST api/my","user":"de77f9e5-3895-11ec-b53c-f09e4ac3134cser","level":"WARN","logger":"com.office.cloud.service.impl.PubSubServiceImpl","msg":"Failed publishing message for : some_name to pub sub: com.google.api.gax.rpc.NotFoundException: io.grpc.StatusRuntimeException: NOT_FOUND: Resource not found (resource=emulator--management-topic)."}
{"ts":"10 29 2021 14:24:47.541","th":"http-nio-9700-exec-1","request":"","":"some_name","api":"POST api/my","user":"de793266-3895-11ec-b53c-f09e4ac3134cser","level":"WARN","logger":"com.office.cloud.exception.ErrorHandler","msg":"Internal error. com.google.api.gax.rpc.NotFoundException: io.grpc.StatusRuntimeException: NOT_FOUND: Resource not found (resource=emulator--management-topic).. 
com.office.cloud.exception.PayrollException: null
	-- work stuff omitted --
	at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[?:1.8.0_301]
	at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) ~[?:1.8.0_301]
	at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[?:1.8.0_301]
	at java.lang.reflect.Method.invoke(Method.java:498) ~[?:1.8.0_301]
	at org.springframework.web.method.support.InvocableHandlerMethod.doInvoke(InvocableHandlerMethod.java:197) ~[spring-web-5.3.8.jar:5.3.8]
	at org.springframework.web.method.support.InvocableHandlerMethod.invokeForRequest(InvocableHandlerMethod.java:141) ~[spring-web-5.3.8.jar:5.3.8]
	at org.springframework.web.servlet.mvc.method.annotation.ServletInvocableHandlerMethod.invokeAndHandle(ServletInvocableHandlerMethod.java:106) ~[spring-webmvc-5.3.8.jar:5.3.8]
	at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.invokeHandlerMethod(RequestMappingHandlerAdapter.java:894) ~[spring-webmvc-5.3.8.jar:5.3.8]
	at org.springframework.web.servlet.mvc.method.annotation.RequestMappingHandlerAdapter.handleInternal(RequestMappingHandlerAdapter.java:808) ~[spring-webmvc-5.3.8.jar:5.3.8]
	at org.springframework.web.servlet.mvc.method.AbstractHandlerMethodAdapter.handle(AbstractHandlerMethodAdapter.java:87) ~[spring-webmvc-5.3.8.jar:5.3.8]

[lesv]

@beena-yatin-kanyal Are you using the APIary client or the GAPIC client? They are very different. At this point, it's best to be using GAPIC and it has full support for the PubSub emulator. https://github.com/GoogleAPIs/java-pubsub

Our spring boot extensions support the GAPIC Pub Sub library.

This repository is for APIary clients.

Additional material:

• https://developers.google.com/learn/pathways/springboot-google-cloud
• https://spring.io/projects/spring-cloud-gcp
• https://cloud.google.com/blog/products/application-development/modernize-your-java-apps-with-spring-cloud-gcp

[beena-yatin-kanyal]

Hello @lesv. I hope your vacation was great.

I am not sure if I understand your answer correctly. we are using the below library to publish/subscribe messages from the real-time GCP pub/sub however the same code breaks (on calling .get()) if I am using the pub/sub emulator with the error snippet mentioned above.

implementation ('com.google.cloud:spring-cloud-gcp-pubsub') {
        exclude group: 'javax.annotation', module: 'javax.annotation-api'
}

Do you want me to change this to something else? Probably spring-cloud-gcp-starter-pubsub? The current scenario is like -
[1] - start the emulator
[2] - use the python scripts to create topic + subscriptions
[3] - modify the spring boot gcloud config class to listen to the emulator running on localhost
[4] - use the app (** breaks here **)

would appreciate it if this can be reopened and addressed.

[beena-yatin-kanyal]

@lesv - can we please have this reopened or if maybe anyone can help? this is impacting development guidelines,

Note - remove the .get() method the message is pushed to the topic (I believe) but not received on the push endpoint.

[beena-yatin-kanyal]

@lesv much thanks. can you or the google development team take a look at it and suggest why the pub/sub emulator is not able to send the message to the push-endpoint? Looking forward to the help.

[beena-yatin-kanyal]

this can be closed now as I have resolved the issue. thanks. To get the pub/sub working on the emulator set the publisher builder with the mock channel provider and credential provider values. Snippet below -

@Bean
@Qualifier(value = "publisher1")
public Publisher publisher1(RetrySettings retrySettings) throws IOException {
TopicName topicName = TopicName.of(projectId, someTopic);
final Publisher.Builder builder = Publisher.newBuilder(topicName)
	.setRetrySettings(retrySettings);
if (usePubSubEmulator) {
  log.info("Setting the {} on pub/sub emulator", someTopic);
  builder.setChannelProvider(transportChannelProvider())
	  .setCredentialsProvider(credentialsProvider());
}
return builder.build();
}

In my opinion, pub/sub emulator documentation needs to be updated.