Receive message from Google PubSub continuously using Asynchronous Pull

Question

With the help of this link, I've managed to create a small Java application that pulls published messages for one minute. My implementation looks something like this.

    public static void eventListener() throws InterruptedException {

    MessageReceiver receiver = new MessageReceiver() {
      @Override
      public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
        System.out.println("Received message: " + message.getData().toStringUtf8());
        consumer.ack();
      }
    };
    //Subscriber subscriber = null;
    try {
      subscriber = Subscriber.newBuilder(subscription, receiver)
          .setCredentialsProvider(FixedCredentialsProvider.create(creds)).build();
      subscriber.addListener(new Subscriber.Listener() {
        @Override
        public void failed(Subscriber.State from, Throwable failure) {
          // Handle failure. This is called when the Subscriber encountered a fatal error
          // and is
          // shutting down.
          System.err.println(failure);
        }
      }, MoreExecutors.directExecutor());
      subscriber.startAsync().awaitRunning();

      // In this example, we will pull messages for one minute (60,000ms) then stop.
      // In a real application, this sleep-then-stop is not necessary.
      // Simply call stopAsync().awaitTerminated() when the server is shutting down,
      // etc.
      Thread.sleep(60000);
    } finally {
      if (subscriber != null) {
        subscriber.stopAsync().awaitTerminated();
      }
    }
  }

When I call this method in main

public static void main(String[] args) throws InterruptedException {
  eventListener();

}

and upload an object to my Google cloud storage, the program prints a message sent by the publisher, like this

Received message: {
  "kind": "storage#object",
  "id": "roshanbucket/stones.jpg/1553765105996166",
  "selfLink": "https://www.googleapis.com/storage/v1/b/roshanbucket/o/stones.jpg",
  "name": "stones.jpg",
  "bucket": "roshanbucket",
  "generation": "1553765105996166",
  "metageneration": "1",
  "contentType": "image/jpeg",
  "timeCreated": "2019-03-28T09:25:05.995Z",
  "updated": "2019-03-28T09:25:05.995Z",
  "storageClass": "STANDARD",
  "timeStorageClassUpdated": "2019-03-28T09:25:05.995Z",
  "size": "137256",
  "md5Hash": "1GmpUnGeiW+/KU+0U8c8Wg==",
  "mediaLink": "https://www.googleapis.com/download/storage/v1/b/roshanbucket/o/stones.jpg?generation=1553765105996166&alt=media",
  "crc32c": "FMaEGg==",
  "etag": "CIaj1InCpOECEAE="
}

For one minute since the program execution, it prints all the messages received on the account of the object upload, then it stops. To receive the event messages after a minute, I need to restart the application. Now, what I'd like to do is to run the listener continuously, So, I tried to run the method eventListener() inside an infinite loop inside the main method, like this

    public static void main(String[] args) throws InterruptedException {
    while(true) {
      eventListener();
    }
  }

With this, I seem to be able to receive the event message right after every upload, regardless of when I upload the object. But then, every once in a while, it throws this stack trace.

    Mar 28, 2019 12:56:34 PM io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference cleanQueue
SEVERE: *~*~*~ Channel ManagedChannelImpl{logId=6, target=pubsub.googleapis.com:443} was not shutdown properly!!! ~*~*~*
    Make sure to call shutdown()/shutdownNow() and wait until awaitTermination() returns true.
java.lang.RuntimeException: ManagedChannel allocation site
    at io.grpc.internal.ManagedChannelOrphanWrapper$ManagedChannelReference.<init>(ManagedChannelOrphanWrapper.java:103)
    at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:53)
    at io.grpc.internal.ManagedChannelOrphanWrapper.<init>(ManagedChannelOrphanWrapper.java:44)
    at io.grpc.internal.AbstractManagedChannelImplBuilder.build(AbstractManagedChannelImplBuilder.java:440)
    at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createSingleChannel(InstantiatingGrpcChannelProvider.java:223)
    at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.createChannel(InstantiatingGrpcChannelProvider.java:164)
    at com.google.api.gax.grpc.InstantiatingGrpcChannelProvider.getTransportChannel(InstantiatingGrpcChannelProvider.java:156)
    at com.google.api.gax.rpc.ClientContext.create(ClientContext.java:157)
    at com.google.cloud.pubsub.v1.stub.GrpcSubscriberStub.create(GrpcSubscriberStub.java:260)
    at com.google.cloud.pubsub.v1.Subscriber.doStart(Subscriber.java:268)
    at com.google.api.core.AbstractApiService$InnerService.doStart(AbstractApiService.java:148)
    at com.google.common.util.concurrent.AbstractService.startAsync(AbstractService.java:225)
    at com.google.api.core.AbstractApiService.startAsync(AbstractApiService.java:120)
    at com.google.cloud.pubsub.v1.Subscriber.startAsync(Subscriber.java:260)
    at listener.AsynchronousPull.eventListener(AsynchronousPull.java:57)
    at listener.AsynchronousPull.main(AsynchronousPull.java:74)

But, it still prints the message after every upload, while throwing the stack trace every now and then. I don't have much experience with threads and I'd really appreciate some help with fixing this issue.

Answer 1

Calling eventListener() in a tight loop is not what you want to do here. This is going to create many new instances of a subscriber that receive messages that each live for 60 seconds. What you want is for the single instance of subscriber you create to live until a time at which you want to shut it down. Generally, you would do this by creating the subscriber and awaiting its termination via awaitTerminated().

The code above would be altered to be like this:

public static void eventListener() throws InterruptedException {
  MessageReceiver receiver = new MessageReceiver() {
    @Override
    public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
      System.out.println("Received message: " + message.getData().toStringUtf8());
      consumer.ack();
    }
  };
  Subscriber subscriber = null;
  try {
    subscriber = Subscriber.newBuilder(subscription, receiver)
        .setCredentialsProvider(FixedCredentialsProvider.create(creds)).build();
    subscriber.addListener(new Subscriber.Listener() {
      @Override
      public void failed(Subscriber.State from, Throwable failure) {
        // Handle failure. This is called when the Subscriber encountered a fatal error
        // and is
        // shutting down.
        System.err.println(failure);
      }
    }, MoreExecutors.directExecutor());
    subscriber.startAsync().awaitRunning();
    subscriber.awaitTerminated();
  } finally {
    if (subscriber != null) {
      subscriber.stopAsync().awaitTerminated();
    }
  }
}

public static void main(String[] args) throws InterruptedException {
  eventListener();
}

If you just want the subscriber to stop when the application terminates and don't want to bother with any additional cleanup, then the code above will work, allowing the subscriber to run and receive messages until an error occurs or the application is shut down. If you want to do some cleanup on clean termination of the application, e.g., you want to ensure that any messages already being handled by receiveMessage run to completion, then you can attach a shutdown hook to catch such terminations (though it will not run in all circumstances). In this hook, you would call stopAsync(). For example, you could insert the following before the try block:

Runtime.getRuntime().addShutdownHook(new Thread() {
    public void run() {
      subscriber.stopAsync().awaitTerminated();
    }
});