Google-cloud-dotnet: Latency issues with Pub / Sub service

[kingerson]

Hello everyone, I am using the GCP Pub / Sub service, but I am having a problem using it, since in 15 minutes it is only able to process the sending of 20 thousand messages and I would like to know why this is happening ... yes This service is supposed to process millions of events per second, the following lines of code are what I use to be able to send the messages.

This method call my function.

private async Task ProcessNotification(IEnumerable<NotificationPush> notifications)
        {
            await notifications.ParallelForEachAsync(async m => {
                var notification_id = GenerateNotificationId();
                var sent_date = _dateService.GetDate();
                var notification_message = new NotificationMessage
                {
                    rp_notification_id = notification_id,
                    rp_sequential_id = m.id_sequencial,
                    register_date = sent_date,
                    message = m.notification.body,
                    uuid = m.uuid,
                    application = m.application_code,
                    rp_campaign_id = m.campaign_id,
                    sent_date = sent_date
                };

                try
                {
                    await _pubSubService.PublishToTopic(notification_message);
                }
                catch (Exception)
                {

                    throw;
                }
            }, maxDegreeOfParallelism: 200);
        }

This implementation method.

public async Task PublishToTopic<T>(T model)
        {
            var topicName = new TopicName(_configuration["GCP_ProjectId"], _configuration["GCP_TopicId"]);
            var _publisherClient = await CreateClient();

            var json = JsonConvert.SerializeObject(model, Formatting.None);

            var data = string.Join("\n", json);

            var message = new PubsubMessage()
            {
                Data = ByteString.CopyFromUtf8(data)
            };

            var response = await _publisherClient.PublishAsync(topicName, new List<PubsubMessage> { message });
        }

This my connection method.

public async Task<PublisherClient> CreateClient(TopicName topicName)
        {
                var isWebApp = _configuration["isWebApp"];
                var binDirectory = Path.GetDirectoryName(Assembly.GetExecutingAssembly().Location);
                var rootDirectory = Path.GetFullPath(Path.Combine(binDirectory, ".."));

                var config = Path.Combine(rootDirectory, "authgcp.json");

                GoogleCredential credential = null;
                using (var jsonStream = new FileStream(config, FileMode.Open, FileAccess.Read, FileShare.Read))
                    credential = GoogleCredential.FromStream(jsonStream);


                PublisherServiceApiClient publisherService = await new PublisherServiceApiClientBuilder { ChannelCredentials = credential.ToChannelCredentials() }.BuildAsync();


                _client = publisherService;
            return _client;
        }

[jskeet]

Hopefully Chris will be able to have a look at this for you as our resident Pub/Sub expect, but as a first pass, this looks odd to me:

var data = string.Join("\n", json);

json is just a single string - so what you attempting to do with the string.Join call?

Separately, it looks like you're using the "raw" API rather than the high-performance clients that wrap them specifically to scale better (PublisherClient in this case). That means you're issuing a separate RPC per message, whereas I believe PublisherClient batches them for much greater performance. Is there any reason you're doing that?

[chrisdunelm]

Is @jskeet suggests, please use the high-performance PublisherClient, as this batches messages and uses multiple network connections to allow higher through-put.
And if you are sending large messages (roughly a few hundred bytes or more), there may be a bandwidth-related performance hit too.

[kingerson]

Hi @chrisdunelm @jskeet I am using PublisherServiceApiClient because it runs on serverless (Azure function)
references Status(StatusCode=DeadlineExceeded, Detail="Deadline Exceeded") #5346

[jskeet]

Okay, it would have been really useful to give that context to start with. Chris will have more details, but as you're working through a batch of messages (notifications) I'd suggest trying to publish batches of messages rather than using one RPC per message.

[kingerson]

@jskeet I'm going to try to send it in batches as you suggest, but I would like to know what is the limit of messages that can be sent in the following code ...

await _publisherClient.PublishAsync(topicName, new List<PubsubMessage> { message });

[jskeet]

I would suggest using PublisherClient.ApiMaxBatchingSessions to determine the maximum number - note that there are two maximum values here; the number of messages (1000) and the total size in bytes (1,000,000 bytes). If you know your messages will be small (less than 1,000 bytes each) you may well be able to just use 1,000 message batches.

[kingerson]

Hi @jskeet , thank you so much for the answer, we will do the test and I comment how it goes.

[kingerson]

Hi @jskeet

I have the following scenario

Azure Function to Pubsub service (publish batches of messages - total less 1,000,000 bytes)
then a Cloud Function listens and sends to biqquery in streaming mode (In the cloud function there are no crashes)

But there are messages that are being lost and I don't know why

[jskeet]

Does the Azure Function complete the publish RPC successfully before the function itself completes? If you can provide a complete example that I could deploy to Azure and reproduce the issue, that would make it easier to help you. It would also be easier if we could only use functions for one side of things:

• If you publish via a simple console app on your developer box, does the Cloud Function manage to consume all the messages without loss?
• If you publish via Azure functions, but then subscribe via a simple console app on your developer box, does that work without loss?

We really need to use divide and conquer to work out what's going on.

[kingerson]

Hello @jskeet @chrisdunelm we have detected that the problem not only occurs in Azure Function but also in a console project locally.

The case is the following: the program downloads a file from the blob storage and deserializes 200K records, then sends it to the pub / sub service and is received by a cloud function trigger that consequently sends the information to BigQuery in streaming mode. The problem is that of those 200K records only approximately 190K are inserted (the number is never the same sometimes 191K, 192K, 197K), Initially we thought it was a problem with the Cloud Function, but we saw in the logs that there were no errors (crash | critical)

I am attaching the console project in .NET Core and the cloud function in python so that you can emulate the problem. If you need anything else please let me know.

repo :

https://github.com/kingerson/send-data-pub-sub

Thank you

[jskeet]

Okay, it's good to know it can be reproduced locally.

When you write:

Initially we thought it was a problem with the Cloud Function, but we saw in the logs that there were no errors

... that doesn't mean the problem is definitely not in the Cloud Function. (There could be a bug in the Python PubSub library, for example - I'm certainly not saying that's the case, just saying it's possible.)

Have you tried reproducing this with two simple console applications (not downloading a file or anything, just generating records)? The simpler we can make it to reproduce this, the better. I'm happy to try to take your current code and produce something more standalone.

[kingerson]

@jskeet
Yes, the test was also done by generating records in the console application ... having the same issue.

Now I have added another method in my repository generating records locally.

https://github.com/kingerson/send-data-pub-sub

[jskeet]

Great, thanks. If you don't have a .NET console app that consumes the messages handy, that's fine - I can write one. (I'll have to see whether I get a chance to do this before the end of the week... it may be next week.)

[jskeet]

Okay, I've now created my own console app to generate and publish 1000 batches of 200 messages, and another console app to pull those messages - which completed (taking rather longer, but it was a single-threaded pull model) and pulled all 200,000 messages.

I'll now try a slightly different tack: a C# Cloud Function that inserts a row in BigQuery like your code does.

[jskeet]

I'm afraid I still can't reproduce this. I've created a console app and a function: the console app creates a topic, waits for you to hit return, then publishes 200,000 messages to the topic in 1000 batches of 200. Each message contains data that's just a string of the form "Batch {batch} message {message}", e.g. "Batch 23 message 30".

The function inserts a row into a BigQuery table with two values in the row:

• The source of the PubSub event (so basically the topic name)
• The text extracted from the event

(You need to create the BigQuery table first, with a trivial schema - two string columns, called "text" and "source".)

Then in the BigQuery console, just query SELECT source, count(1) as count FROM YOUR_TABLE_NAME_HERE group by source - if everything goes well, you should see a count of 200,000, not long after the publisher completes.

Publisher app:

using Dasync.Collections;
using Google.Cloud.PubSub.V1;
using Google.Protobuf;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace TestPublisher
{
    class Program
    {
        private const int MessagesPerBatch = 200;
        private const int Batches = 1000;

        private static async Task Main(string[] args)
        {
            if (args.Length != 2)
            {
                Console.WriteLine($"Please specify the project ID and topic ID");
                return;
            }

            string projectId = args[0];
            string topicId = args[1];

            var client = PublisherServiceApiClient.Create();
            var topicName = new TopicName(projectId, topicId);
            await client.CreateTopicAsync(topicName);

            Console.WriteLine("Created topic. Please hit return to start generating messages...");
            Console.ReadLine();

            var messages = GenerateMessageBatches().ToList();

            Console.WriteLine("Publishing messages...");
            await messages.ParallelForEachAsync(async batch =>
            {
                try
                {
                    await PublishBatch(batch);
                    Console.Write(".");
                }
                catch (Exception e)
                {
                    Console.WriteLine($"Error: {e}");
                }

            }, maxDegreeOfParallelism: 0);
            Console.WriteLine();
            Console.WriteLine("Publishing complete");

            async Task PublishBatch(IEnumerable<string> messages)
            {
                var pubsubMessages = messages
                    .Select(text => new PubsubMessage { Data = ByteString.CopyFromUtf8(text) })
                    .ToList();

                await client.PublishAsync(topicName, pubsubMessages);
            }
        }

        private static IEnumerable<IEnumerable<string>> GenerateMessageBatches()
        {
            for (int batch = 0; batch < Batches; batch++)
            {
                yield return Enumerable.Range(0, MessagesPerBatch)
                    .Select(index => $"Batch {batch} message {index}")
                    .ToList();
            }
        }
    }
}

Function:

using CloudNative.CloudEvents;
using Google.Api.Gax;
using Google.Cloud.BigQuery.V2;
using Google.Cloud.Functions.Framework;
using Google.Cloud.Functions.Hosting;
using Google.Events.Protobuf.Cloud.PubSub.V1;
using Microsoft.AspNetCore.Hosting;
using Microsoft.Extensions.DependencyInjection;
using System.Threading;
using System.Threading.Tasks;

[assembly: FunctionsStartup(typeof(TestCloudFunction.Startup))]

namespace TestCloudFunction
{
    public class Startup : FunctionsStartup
    {
        public override void ConfigureServices(WebHostBuilderContext context, IServiceCollection services) =>
            services.AddSingleton(BigQueryClient.Create(Platform.Instance().ProjectId));
    }

    public class Function : ICloudEventFunction<MessagePublishedData>
    {
        // Adjust these appropriately, or take them from configuration...
        private const string DataSetId = "issue5403";
        private const string TableId = "testtable";

        private readonly BigQueryClient _bqClient;

        public Function(BigQueryClient bqClient) => _bqClient = bqClient;

        public async Task HandleAsync(CloudEvent cloudEvent, MessagePublishedData data, CancellationToken cancellationToken)
        {
            var row = new BigQueryInsertRow
            {
                ["text"] = data.Message.TextData,
                ["source"] = cloudEvent.Source.ToString()
            };
            await _bqClient.InsertRowAsync(DataSetId, TableId, row);
        }
    }
}

If you could reproduce my findings to start with, that would be helpful - we can then try to work out what's different in your code...

[kingerson]

Hi @chrisdunelm , thank you so much for the answer, we will do the test and I comment how it goes.

[jskeet]

Any updates on this? We'd like to either have something actionable, or close it until there is something actionable.

[kingerson]

Hello @jskeet , apparently the problem was with the cloud function in python, since we made the change to dotnet and everything worked correctly, sorry for the delay in responding, but we have been doing many stress tests to ensure that everything is working correctly.

[jskeet]

Right, glad to hear it - will close this.