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]

[hannahrogers-google]

To use the emulator with Java, some code modifications are required. See our documentation here: https://cloud.google.com/pubsub/docs/emulator#accessing_environment_variables

Did you follow the example listed, and set the credentialsProvider/channelProvider when creating the Publisher?

[beena-yatin-kanyal]

yes, I followed all the steps listed in the given document. started the emulator first, set the required environment variables, and then the topics/subscription are created via the python scripts. Once done the following was updated in the backend code -

Configuration file

// DEVELOPER - GCLOUD PUB/SUB CHANGES
// START
@Value("${emulator.pubsub.host}")
String host;
@Value("${emulator.pubsub.port}")
int port;

@Bean(name = "credentialsProvider")
@ConditionalOnProperty(value = "emulator.usePubSub", havingValue = "true")
public CredentialsProvider credentialsProvider() {
  return NoCredentialsProvider.create();
}

@Bean(name = "messageChannel")
@ConditionalOnProperty(value = "emulator.usePubSub", havingValue = "true")
public ManagedChannel messageChannel() {
  final String serviceEndpoint = host + ":" + port;
  log.info("Pub/sub emulator :: service-url = {}", serviceEndpoint);
  return ManagedChannelBuilder.forTarget(serviceEndpoint).usePlaintext().build();
}

@Bean(name = "transportChannelProvider")
@ConditionalOnProperty(value = "emulator.usePubSub", havingValue = "true")
@DependsOn(value = {"messageChannel"})
public TransportChannelProvider transportChannelProvider() {
  return FixedTransportChannelProvider.create(GrpcTransportChannel.create(messageChannel()));
}
// END

 @Bean
@Qualifier(value = "qualifier1")
public Publisher qualifier1(RetrySettings retrySettings) throws IOException {
  TopicName topicName = TopicName.of(projectId, topic1);
  return Publisher.newBuilder(topicName).setRetrySettings(retrySettings).build();
}

@Bean
@Qualifier(value = "qualifier2")
public Publisher qualifier2(RetrySettings retrySettings) throws IOException {
  TopicName topicName = TopicName.of(projectId, topic2);
  return Publisher.newBuilder(topicName).setRetrySettings(retrySettings).build();
}

application-local.properties

# PUB/SUB
emulator.pubsub.host=localhost
emulator.pubsub.port=5189
emulator.usePubSub=true
spring.cloud.gcp.pubsub.emulator-host=${emulator.pubsub.host}:${emulator.pubsub.port}

a quick catch here is that if I remove the get() method from the below line the message is sent to the topic (I believe) but not delivered on the configured push endpoint.

return publisher.publish(pubsubMessage).get();

[beena-yatin-kanyal]

@hannahrogers-google - did you get a chance to look at it? anyone who can help? it is impacting our development guidelines.

[elefeint]

@beena-yatin-kanyal The code creates a bean transportChannelProvider, but it does not use it anywhere. Inject transportChannelProvider bean into qualifier1() and qualifier2() and set it on the publisher with Publisher.newBuilder(topicName).setChannelProvider(...)

[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.