Google-cloud-dotnet: Acking in stream sometimes misbehaving

Created on 2 Jun 2020  路  11Comments  路  Source: googleapis/google-cloud-dotnet

Raised as part of #5000

(Code to reproduce when I get time.)

pubsub external needs more info p2 bug
Source
picture jskeet

All 11 comments

Hi Jon, below you'll find the test code. It is loosely based on the #5000 code you supplied. You can use the boolean BULK_ACK to either Ack each message in the stream individually or Ack them all at once after receiving. When BULK_ACK is false it might need a couple of runs to see messages reappear, in some runs it does not happen at all. When BULK_ACK is true I usually (always?) see most (all?) message reappear.

```c#
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Globalization;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Google.Cloud.PubSub.V1;
using Google.Protobuf;

namespace PubSubProblem
{
internal sealed class Program
{
private const string ProjectId = "";

    private const int TotalMessages = 1000;
    private static readonly TimeSpan PublishDelay = TimeSpan.FromMilliseconds(10);

    private static readonly bool BULK_ACK = true;

    private static void debug(string value) => Console.Write($"{DateTime.UtcNow:o}" + "\t- " + value);

    private static async Task Main()
    {
        var publisher = PublisherServiceApiClient.Create();
        var subscriber = SubscriberServiceApiClient.Create();
        var runId = DateTime.UtcNow.ToString("yyyyMMddTHHmmss", CultureInfo.InvariantCulture);
        var topicName = new TopicName(ProjectId, $"issue5012-{runId}");
        var subscriptionName = new SubscriptionName(ProjectId, $"issue5012-{runId}");

        publisher.CreateTopic(topicName);
        subscriber.CreateSubscription(subscriptionName, topicName, pushConfig: null, ackDeadlineSeconds: 30);

        PublishMessages(publisher, topicName);
        await StreamingPullMessages(subscriber, subscriptionName);

        while (true)
            try
            {
                debug("\r\n=== There should not be any messages left by now ===\r\n");
                debug("(Re-)entering endless loop of 'StreamingPullMessages'...\r\n");
                debug("Might take a while (Ack deadline + some random time) to get the messages again.\r\n");
                debug("They might also not come back after all...\r\n");
                await StreamingPullMessages(subscriber, subscriptionName);
            }
            catch
            {
                debug("=== Got some exception. Probably a timeout of the stream. Ignoring... ===\r\n");
            }
    }

    private static void PublishMessages(PublisherServiceApiClient publisher, TopicName topicName)
    {
        for (int index = 1; index <= TotalMessages; index++)
        {
            publisher.Publish(topicName, new[] {new PubsubMessage {Data = ByteString.CopyFromUtf8($"Message {index}")}});
            debug($"Published {index} out of {TotalMessages} messages\r");
            Thread.Sleep(PublishDelay);
        }

        debug("\n");
        debug("Publishing completed\r\n");
    }

    private static async Task StreamingPullMessages(SubscriberServiceApiClient subscriber, SubscriptionName subscriptionName)
    {
        List<string> ackList = new List<string>();

        var stream = subscriber.StreamingPull();
        await stream.WriteAsync(new StreamingPullRequest
        {
            SubscriptionAsSubscriptionName = subscriptionName, ClientId = Guid.NewGuid().ToString(),
            StreamAckDeadlineSeconds = 60
        });
        var responses = stream.GetResponseStream();
        var totalReceived = 0;

        var overallStopwatch = Stopwatch.StartNew();
        while (totalReceived < TotalMessages)
        {
            var stopwatch = Stopwatch.StartNew();
            await responses.MoveNextAsync();
            var messages = responses.Current.ReceivedMessages;
            stopwatch.Reset();

            ackList.AddRange(messages.Select(message => message.AckId));
            if (!BULK_ACK)
            {
                await stream.WriteAsync(new StreamingPullRequest {AckIds = {ackList}});
            }

            totalReceived += messages.Count;
            debug($"Received {messages.Count} message in this iteration and a total of {totalReceived} out of {TotalMessages} messages\r\n");
        }

        if (BULK_ACK)
        {
            debug("Bulk Acknowledging\r\n");
            await stream.WriteAsync(new StreamingPullRequest {AckIds = {ackList}});
        }

        overallStopwatch.Stop();
        debug($"Total time to received {TotalMessages} messages: {overallStopwatch.ElapsedMilliseconds}ms.\r\n");

        // WriteCompleteAsync: 'Completes the stream when all buffered messages have been sent.' but does not change the behaviour.
        // If you leave it here or not does not change the Ack results. Assumed that it would at least gracefully complete the
        // ACK requests
        await stream.WriteCompleteAsync();
    }
}

}


Example output with `BULK_ACK = false`, one message returns once (listed with `>>>`) but seems to be Ack'ed in that iteration and it does not reappear.


2020-06-02T20:52:19.5528178Z    - Published 1000 out of 1000 messages
2020-06-02T20:52:19.5528554Z    - Publishing completed
2020-06-02T20:52:20.9336804Z    - Received 1 message in this iteration and a total of 1 out of 1000 messages
2020-06-02T20:52:20.9340126Z    - Received 1 message in this iteration and a total of 2 out of 1000 messages
2020-06-02T20:52:20.9341845Z    - Received 1 message in this iteration and a total of 3 out of 1000 messages
...
2020-06-02T20:52:25.3325926Z    - Received 1 message in this iteration and a total of 998 out of 1000 messages
2020-06-02T20:52:25.3342078Z    - Received 1 message in this iteration and a total of 999 out of 1000 messages
2020-06-02T20:52:25.3565331Z    - Received 1 message in this iteration and a total of 1000 out of 1000 messages
2020-06-02T20:52:25.3566145Z    - Total time to received 1000 messages: 5769ms.
2020-06-02T20:52:25.3591018Z    - 
=== There should not be any messages left by now ===
2020-06-02T20:52:25.3591179Z    - (Re-)entering endless loop of 'StreamingPullMessages'...
2020-06-02T20:52:25.3591272Z    - Might take a while (Ack deadline + some random time) to get the messages again.
2020-06-02T20:52:25.3591338Z    - They might also not come back after all...

2020-06-02T20:53:23.1612409Z - Received 1 message in this iteration and a total of 1 out of 1000 messages
2020-06-02T20:55:03.3939721Z - === Got some exception. Probably a timeout of the stream. Ignoring... ===
2020-06-02T20:55:03.3940144Z -
=== There should not be any messages left by now ===
2020-06-02T20:55:03.3940215Z - (Re-)entering endless loop of 'StreamingPullMessages'...
2020-06-02T20:55:03.3940246Z - Might take a while (Ack deadline + some random time) to get the messages again.
2020-06-02T20:55:03.3940276Z - They might also not come back after all...
2020-06-02T20:56:41.1929075Z - === Got some exception. Probably a timeout of the stream. Ignoring... ===
2020-06-02T20:56:41.1929443Z -
=== There should not be any messages left by now ===
2020-06-02T20:56:41.1929513Z - (Re-)entering endless loop of 'StreamingPullMessages'...
2020-06-02T20:56:41.1929575Z - Might take a while (Ack deadline + some random time) to get the messages again.
2020-06-02T20:56:41.1929644Z - They might also not come back after all...
2020-06-02T20:57:45.6093590Z - === Got some exception. Probably a timeout of the stream. Ignoring... ===


Example output with `BULK_ACK = true`. In the loop all messages appear again. It does not seem to Acknowledge any of them. Sometimes only some of the messages reappear before timeout.


2020-06-02T21:06:52.1657613Z    - Published 1000 out of 1000 messages
2020-06-02T21:06:52.1657908Z    - Publishing completed
2020-06-02T21:06:53.5319156Z    - Received 1 message in this iteration and a total of 1 out of 1000 messages
2020-06-02T21:06:53.5322099Z    - Received 1 message in this iteration and a total of 2 out of 1000 messages
2020-06-02T21:06:53.5323312Z    - Received 1 message in this iteration and a total of 3 out of 1000 messages
...
2020-06-02T21:06:54.2322750Z    - Received 1 message in this iteration and a total of 997 out of 1000 messages
2020-06-02T21:06:54.2323067Z    - Received 1 message in this iteration and a total of 998 out of 1000 messages
2020-06-02T21:06:54.2329313Z    - Received 1 message in this iteration and a total of 999 out of 1000 messages
2020-06-02T21:06:54.2329970Z    - Received 1 message in this iteration and a total of 1000 out of 1000 messages
2020-06-02T21:06:54.2330040Z    - Bulk Acknowledging
2020-06-02T21:06:54.2456610Z    - Total time to received 1000 messages: 2051ms.
2020-06-02T21:06:54.2491748Z    - 
=== There should not be any messages left by now ===
2020-06-02T21:06:54.2491962Z    - (Re-)entering endless loop of 'StreamingPullMessages'...
2020-06-02T21:06:54.2492037Z    - Might take a while (Ack deadline + some random time) to get the messages again.
2020-06-02T21:06:54.2492091Z    - They might also not come back after all...

2020-06-02T21:06:55.2957445Z - Received 1 message in this iteration and a total of 1 out of 1000 messages
2020-06-02T21:06:55.2958220Z - Received 1 message in this iteration and a total of 2 out of 1000 messages
2020-06-02T21:06:55.2958796Z - Received 1 message in this iteration and a total of 3 out of 1000 messages
...
2020-06-02T21:08:19.4880334Z - Received 1 message in this iteration and a total of 998 out of 1000 messages
2020-06-02T21:08:19.4880831Z - Received 1 message in this iteration and a total of 999 out of 1000 messages
2020-06-02T21:08:19.4969370Z - Received 1 message in this iteration and a total of 1000 out of 1000 messages
2020-06-02T21:08:19.4969525Z - Bulk Acknowledging
2020-06-02T21:08:19.5082811Z - Total time to received 1000 messages: 85257ms.
2020-06-02T21:08:19.5085253Z -
=== There should not be any messages left by now ===
2020-06-02T21:08:19.5085424Z - (Re-)entering endless loop of 'StreamingPullMessages'...
2020-06-02T21:08:19.5085529Z - Might take a while (Ack deadline + some random time) to get the messages again.
2020-06-02T21:08:19.5085619Z - They might also not come back after all...
2020-06-02T21:08:19.9991209Z - Received 1 message in this iteration and a total of 1 out of 1000 messages
2020-06-02T21:08:19.9992595Z - Received 1 message in this iteration and a total of 2 out of 1000 messages
2020-06-02T21:08:19.9993192Z - Received 1 message in this iteration and a total of 3 out of 1000 messages
...
2020-06-02T21:09:20.2181825Z - Received 1 message in this iteration and a total of 998 out of 1000 messages
2020-06-02T21:09:20.2208046Z - Received 1 message in this iteration and a total of 999 out of 1000 messages
2020-06-02T21:09:20.2215341Z - Received 1 message in this iteration and a total of 1000 out of 1000 messages
2020-06-02T21:09:20.2215487Z - Bulk Acknowledging
2020-06-02T21:09:20.2296464Z - Total time to received 1000 messages: 60720ms.
2020-06-02T21:09:20.2297582Z -
=== There should not be any messages left by now ===
2020-06-02T21:09:20.2297733Z - (Re-)entering endless loop of 'StreamingPullMessages'...
2020-06-02T21:09:20.2297813Z - Might take a while (Ack deadline + some random time) to get the messages again.
2020-06-02T21:09:20.2297875Z - They might also not come back after all...
2020-06-02T21:09:21.1351542Z - Received 1 message in this iteration and a total of 1 out of 1000 messages
2020-06-02T21:09:21.1352487Z - Received 1 message in this iteration and a total of 2 out of 1000 messages
2020-06-02T21:09:21.1354072Z - Received 1 message in this iteration and a total of 3 out of 1000 messages
...
2020-06-02T21:09:21.1733772Z - Received 1 message in this iteration and a total of 174 out of 1000 messages
2020-06-02T21:09:21.1734239Z - Received 1 message in this iteration and a total of 175 out of 1000 messages
2020-06-02T21:09:21.1734735Z - Received 1 message in this iteration and a total of 176 out of 1000 messages
2020-06-02T20:56:41.1929075Z - === Got some exception. Probably a timeout of the stream. Ignoring... ===
2020-06-02T20:56:41.1929443Z -
```

Chasx picture Chasx on 2 Jun 2020

The Google Monitoring images of above tests indeed show that BULK_ACK did not acknowledge any message at all at the first iteration (the next iteration does not run because the Ack part is outside the loop and is not reached because of the timeout exception). The non BULK_ACK does Ack although not all after the first iteration.

Individual Ack:
singlemessag

Bulk Ack:
bulkack

Chasx picture Chasx on 2 Jun 2020

Some additional information:

Changing

```c#
if (BULK_ACK)
{
debug("Bulk Acknowledging\r\n");
await stream.WriteAsync(new StreamingPullRequest {AckIds = {ackList}});
}


to

```c#
if (BULK_ACK)
{
    debug("Bulk Acknowledging\r\n");
    await subscriber.AcknowledgeAsync(subscriptionName, ackList);
}

does also leave me regularly with not all messages Ack'ed. It took me 4 restarts of the program but on the 4th run it came back with 208 reappearing messages. It is strange that there is only a couple of seconds between the first iteration receiving them and the second. I'd assume at least 'ack timeout' seconds in between before PubSub would have offered the messages again..?

    2020-06-02T22:40:22.4125230Z    - Published 1000 out of 1000 messages
    2020-06-02T22:40:22.4125936Z    - Publishing completed
    2020-06-02T22:40:23.8830793Z    - Received 1 message in this iteration and a total of 1 out of 1000 messages
    2020-06-02T22:40:23.8833471Z    - Received 1 message in this iteration and a total of 2 out of 1000 messages
    2020-06-02T22:40:23.8834378Z    - Received 1 message in this iteration and a total of 3 out of 1000 messages
    ...
    2020-06-02T22:40:24.5884117Z    - Received 1 message in this iteration and a total of 998 out of 1000 messages
    2020-06-02T22:40:24.5884589Z    - Received 1 message in this iteration and a total of 999 out of 1000 messages
    2020-06-02T22:40:24.5885149Z    - Received 1 message in this iteration and a total of 1000 out of 1000 messages
    2020-06-02T22:40:24.5885244Z    - Bulk Acknowledging
    2020-06-02T22:40:24.6548755Z    - Total time to received 1000 messages: 2204ms.
    2020-06-02T22:40:24.6572427Z    - 
    === There should not be any messages left by now ===
    2020-06-02T22:40:24.6572575Z    - (Re-)entering endless loop of 'StreamingPullMessages'...
    2020-06-02T22:40:24.6572609Z    - Might take a while (Ack deadline + some random time) to get the messages again.
    2020-06-02T22:40:24.6572636Z    - They might also not come back after all...
>>> 2020-06-02T22:40:25.2834360Z    - Received 1 message in this iteration and a total of 1 out of 1000 messages
>>> 2020-06-02T22:40:25.2835543Z    - Received 1 message in this iteration and a total of 2 out of 1000 messages
>>> 2020-06-02T22:40:25.2836478Z    - Received 1 message in this iteration and a total of 3 out of 1000 messages
>>> ...
>>> 2020-06-02T22:40:25.5836443Z    - Received 1 message in this iteration and a total of 206 out of 1000 messages
>>> 2020-06-02T22:40:25.5836754Z    - Received 1 message in this iteration and a total of 207 out of 1000 messages
>>> 2020-06-02T22:40:25.5837093Z    - Received 1 message in this iteration and a total of 208 out of 1000 messages
    2020-06-02T22:41:34.2214209Z    - === Got some exception. Probably a timeout of the stream. Ignoring... ===
    2020-06-02T22:41:34.2214499Z    -
Chasx picture Chasx on 3 Jun 2020

Thanks for the code and results. I probably won't get a chance to look in detail this week, but I'll get to it when I can. Given that these are the "raw" gRPC clients we're using, this is likely to be some subtlety of PubSub service behavior rather than something we can change in the client library, but I'll look into it.

jskeet picture jskeet on 3 Jun 2020

Finally got round to working on this... some notes as I go along.

  • I've seen mixed results with smaller numbers - only publishing 10 messages and bulk acking them, I had one run where everything worked, and one run where I saw the same 10 again and again
  • If I publish 1000 messages but "bulk ack in batches of 100" then I consistently see 100 afterwards. I don't (yet) know if that's the first or the last 100.
jskeet picture jskeet on 17 Jun 2020

Next test: acking 5 messages at a time out of 10. Sometimes this works with no problems (no messages left after the first round); other times, it always leaves "the messages we received last in the first round".

Ah, but if the call times out while waiting to get more messages, we won't bulk ack the ones we did.
Time to change the while (true) loop to stop after waiting for a period, perhaps (before the exception is thrown)

jskeet picture jskeet on 17 Jun 2020

I still have no discernible pattern here. But acking 5 messages at a time, with 50 messages in total, seems to reproduce the problem fairly reliably.
Will report internally and see where we want to go next.

jskeet picture jskeet on 17 Jun 2020
👍1

Created an issue internally with the following repro description:

  • Create a new topic and a corresponding subscription
  • Publish 1000 messages to the topic separately, with a delay of 10ms between each message
  • Start a streaming pull
  • Pull all the messages (this only takes a couple of seconds)
  • Ack all the messages in 10 batches of 100 (via StreamingPullRequest calls with AckIds, in the same stream)
  • "Complete" the call (in .NET, this is via stream.WriteCompleteAsync; I don't know what it's called in other languages.)
  • Look at the metrics for the topic in Cloud Monitoring

Every time, you end up with 100 messages that claim not to be acked. It's as if the last ACK request is ignored. If you pull the topic again, the last 100 messages are replayed.

jskeet picture jskeet on 18 Jun 2020
👍1

Okay, I've heard back from the team. This is a known issue with ACKing in the stream, and in fact the high performance wrapper classes don't do that - they call Acknowledge or AcknowledgeAsync to acknowledge the requests separately from the stream. For example:

while (ackList.Count > 0)
{
    var count = Math.Min(ackList.Count, BulkAcknowledgeBatchSize);
    var request = new AcknowledgeRequest
    {
        SubscriptionAsSubscriptionName = subscriptionName,
        AckIds = { ackList.Take(count) }
    };
    Log($"Bulk Acknowledging {request.AckIds.Count} messages\r\n");
    await subscriber.AcknowledgeAsync(request);
    ackList.RemoveRange(0, count);
}

I've been acking 100 messages per request like that, and everything seems to get ACKed that way.

The team also still recommends using the wrapper classes if you possibly can, by the way - I realize you have reasons to use the lower-level API.

If you're happy with this workaround, I'll close the issue - I don't think there's anything more we can do within the library, and it seems a reasonable workaround.

jskeet picture jskeet on 26 Jun 2020

(Sorry, hadn't meant to close it immediately!)

jskeet picture jskeet on 26 Jun 2020

Closing on the assumption that you're happy enough with the workaround.

jskeet picture jskeet on 8 Jul 2020
Was this page helpful?
0 / 5 - 0 ratings

Related issues

mahesh-bennie picture mahesh-bennie  路  7Comments
riiight picture riiight  路  9Comments
Talento90 picture Talento90  路  6Comments
Eldar-Ahmadov picture Eldar-Ahmadov  路  6Comments
AliSheikhTaheri picture AliSheikhTaheri  路  6Comments