CODEWITHSUNDEEP
shellgoogle-cloud-platformgoogle-cloud-pubsubgoogle-cloud-sdk
Ralph
Ralph

Reputation: 11

Pub Sub Messages still in queue but not pulled

I have a simple shell script that connect to GCP and try to pull Pub/Sub messages from a topic.

When launched, it check if any message exist, does a simple action if so, then ack the message and loop .

It looks like that :

while [ 1 ]
do
   gcloud pubsub subscriptions pull...
   // Do something
   gcloud pubsub subscriptions ack ...
done

Randomly it does not pull the messages : they stay in the queue and are not pulled.

So we tried to add a while loop when getting the message with something like 5 re-try in order to avoid those issues work better but not perfectly. I also think that is a bit shabby...

This issue happened on other project that where migrated from a script shell to Java (for some other reasons) where we used a pull subscription and it work perfectly on those projects now !

We must probably do something wrong but I don't know what...

I have read that sometimes gcloud pull less messages than what's really on the pubsub queue :

https://cloud.google.com/sdk/gcloud/reference/pubsub/subscriptions/pull

But it must at least pull one ... In our case no messages are pulled but randomly.

Is there something to improve here ?

Upvotes: 1

Views: 2404

Answers (1)

Kamal Aboul-Hosn
Kamal Aboul-Hosn

Reputation: 17161

In general, relying on a shell script that uses gcloud to retrieve messages and do something with them is not going to be an efficient way to use Cloud Pub/Sub. It is worth noting that the lack of messages being returned in pull is not indicative of a lack of messages; it just means that messages could not be returned before the pull request's deadline. The gcloud subscriptions pull command sets the returnImmediately property (see info in pull documentation) to true, which basically means that if there aren't messages already quickly accessible in memory, then no messages are going to be returned. This property is deprecated and should not be set to true, so that is probably something that we need to explore changing in gcloud.

You would be better off writing a subscriber using the client libraries that sets up a stream and continuously retrieves messages. If your intention is to run this only periodically, then you could write a job that reads messages and waits some time after messages have not been received and shuts down. Again, this would not guarantee that all messages would be consumed that are available, but it would be true in most cases.

A version of this in Java would look like this:

import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import java.util.concurrent.atomic.AtomicLong;
import org.joda.time.DateTime;

/** A basic Pub/Sub subscriber for purposes of demonstrating use of the API. */
public class Subscriber implements MessageReceiver {
  private final String PROJECT_NAME = "my-project";
  private final String SUBSCRIPTION_NAME = "my-subscription";
  private com.google.cloud.pubsub.v1.Subscriber subscriber;

  private AtomicLong lastReceivedTimestamp = new AtomicLong(0);

  private Subscriber() {

    ProjectSubscriptionName subscription =
        ProjectSubscriptionName.of(PROJECT_NAME, SUBSCRIPTION_NAME);
    com.google.cloud.pubsub.v1.Subscriber.Builder builder =
        com.google.cloud.pubsub.v1.Subscriber.newBuilder(subscription, this);
    try {
      this.subscriber = builder.build();
    } catch (Exception e) {
      System.out.println("Could not create subscriber: " + e);
      System.exit(1);
    }
  }

  @Override
  public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
    // Process message
    lastReceivedTimestamp.set(DateTime.now().getMillis());
    consumer.ack();
  }

  private void run() {
    subscriber.startAsync();
    while (true) {
      long now = DateTime.now().getMillis();
      long currentReceived = lastReceivedTimestamp.get();
      if (currentReceived > 0 && ((now - currentReceived) > 30000)) {
        subscriber.stopAsync();
        break;
      }
      try {
        Thread.sleep(5000);
      } catch (InterruptedException e) {
        System.out.println("Error while waiting for completion: " + e);
      }
    }
    System.out.println("Subscriber has not received message in 30s. Stopping.");
    subscriber.awaitTerminated();
  }

  public static void main(String[] args) {
    Subscriber s = new Subscriber();
    s.run();
    System.exit(0);
  }
}

Upvotes: 1

Related Questions