Weird behavior when pulling messages and using modifyAckDeadline

@eschapira

The scenario is the following: 1 topic and 1 subscription with default ack deadline (10 seconds). I publish 2 messages (message1 and message2) to the topic and then pull messages from the subscription.

After the first pull I set the ack deadline of message1 to a big value (let's say 100 seconds). Then I pull messages from the subscription every 20 seconds. The expected behavior is that pulls return only message2 until ~100 seconds are passed, and then return both message1 and message2.

When I run this scenario against the actual service I get the weird behavior: pull requests made before that ~100 seconds are passed after the deadline modification return no messages (as if both messages where affected by modifyAckDeadline).

The veneer toolkit code that reproduces the error is the following:

PublisherApi publisher = PublisherApi.create(publisherSettings);
SubscriberApi subscriber = SubscriberApi.create(subscriberSettings);
String topicName = PublisherApi.formatTopicName("gcloud-devel", "test-topic");
String subscriptionName =
    SubscriberApi.formatSubscriptionName("gcloud-devel", "test-subscription");
publisher.createTopic(topicName);
subscriber.createSubscription(Subscription.newBuilder()
    .setName(subscriptionName)
    .setTopic(topicName)
    .setAckDeadlineSeconds(10)
    .build());
publisher.publish(topicName, ImmutableList.of(
    PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8("message1")).build(),
    PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8("message2")).build()));
PullResponse pullResponse = subscriber.pull(subscriptionName, true, 2);
List<ReceivedMessage> receivedMessages = pullResponse.getReceivedMessagesList();
for (ReceivedMessage message : pullResponse.getReceivedMessagesList()) {
  System.out.printf("[%d][PULLED MESSAGE %s]%n",
      System.currentTimeMillis(), message.getMessage().getData().toStringUtf8());
}
// Set the deadline of the first message far away
subscriber.modifyAckDeadline(subscriptionName,
    ImmutableList.of(receivedMessages.get(0).getAckId()), 100);

// Second message should be again available for pulling after .sleep()
Thread.sleep(20000);
for (int i = 0; i < 10; i++) {
  pullResponse = subscriber.pull(subscriptionName, true, 2);
  receivedMessages = pullResponse.getReceivedMessagesList();
  if (receivedMessages.isEmpty()) {
    System.out.printf("[%d][PULLED NO MESSAGES]%n", System.currentTimeMillis());
  } else {
    for (ReceivedMessage message : pullResponse.getReceivedMessagesList()) {
      System.out.printf("[%d][PULLED MESSAGE %s]%n",
          System.currentTimeMillis(), message.getMessage().getData().toStringUtf8());
    }
  }
  Thread.sleep(20000);
}
publisher.deleteTopic(topicName);
subscriber.deleteSubscription(subscriptionName);

When I run it against the actual service I get the following output:

[1463750536877][PULLED MESSAGE message1]
[1463750536877][PULLED MESSAGE message2]
[1463750561558][PULLED NO MESSAGES] // <-- here and in the following I would expect message2
[1463750584638][PULLED NO MESSAGES]
[1463750608178][PULLED NO MESSAGES]
[1463750631240][PULLED NO MESSAGES]
[1463750653656][PULLED MESSAGE message1]
[1463750653657][PULLED MESSAGE message2]
[1463750676241][PULLED MESSAGE message1]
[1463750676241][PULLED MESSAGE message2]
[1463750698302][PULLED MESSAGE message1]
[1463750698302][PULLED MESSAGE message2]
[1463750720269][PULLED MESSAGE message1]
[1463750720270][PULLED MESSAGE message2]
[1463750742847][PULLED MESSAGE message1]
[1463750742847][PULLED MESSAGE message2]

Instead when I run it agains the emulator the output is as expected:

[1463749875238][PULLED MESSAGE message1]
[1463749875238][PULLED MESSAGE message2]
[1463749895270][PULLED MESSAGE message2]
[1463749915289][PULLED MESSAGE message2]
[1463749935308][PULLED MESSAGE message2]
[1463749955329][PULLED MESSAGE message2]
[1463749975347][PULLED MESSAGE message1]
[1463749975347][PULLED MESSAGE message2]
[1463749995363][PULLED MESSAGE message1]
[1463749995363][PULLED MESSAGE message2]
[1463750015384][PULLED MESSAGE message2]
[1463750015384][PULLED MESSAGE message1]
[1463750035403][PULLED MESSAGE message2]
[1463750035403][PULLED MESSAGE message1]
[1463750055421][PULLED MESSAGE message2]
[1463750055422][PULLED MESSAGE message1]

@eschapira @garrettjonesgoogle Any idea about what's going on?

I tried this out myself, but adding one thing to the printout, the ack id:

      System.out.printf("[%d][PULLED MESSAGE %s, ackId = %s]%n",
          System.currentTimeMillis(), message.getMessage().getData().toStringUtf8(),
          message.getAckId());

What I discovered is that the ack id was the same for both message1 and message2.

@garrettjonesgoogle

What I discovered is that the ack id was the same for both message1 and message2.

Actually they are very similar but different (they differ of a couple of letters).

I managed to reproduce this issue (thanks to @garrettjonesgoogle) also with gcloud-node (code can be found here).

The problem seems to be related to the fact that the two messages are published using a single call to publish. If we use two publish calls:

publisher.publish(topicName, ImmutableList.of(
    PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8("message1")).build()));
publisher.publish(topicName, ImmutableList.of(
    PubsubMessage.newBuilder().setData(ByteString.copyFromUtf8("message2")).build()));