Pub/Sub: messages stuck in buffer, preventing proper load balancing #17

kir-titievsky · 2018-04-13T21:17:23Z

Repro:

Publish 60 messages, with numbers 1 through 6 as message content. Observe the total backlog for a subscription reach 60 messages and 23 bytes.
Start an instance of a subscriber client, with a single-threaded executor and FlowControl set to 1 message per buffer (see code below). The subscriber takes 10 second to process each message.
Observe that the subscriber processes messages, one at a time, every 10 seconds (see log output below)
The bug: Start two new instances of the same subscriber client, roughly a minute later. Observe: they process no messages. Expected behavior: the two subscribers immediately start processing messages.
Stop the first subscriber client. Observe: the next two subscriber clients start processing messages.

The hypothesis here is that the entire backlog is stuck in the gRPC and other buffers, between the server and the client. So the server thinks the messages are out and being processed, while the client code can't really see the messages. When new clients connect, the server does not have anything to send them. Killing the original client effectively "nacks" the messages in the buffer, by sending a stream close signal to the server. This allows the server to start sending messages to the other clients.

import com.google.api.gax.batching.FlowControlSettings;
import com.google.api.gax.core.InstantiatingExecutorProvider;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import org.threeten.bp.ZonedDateTime;
import org.threeten.bp.format.DateTimeFormatter;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

public class Sub{

    private static AtomicInteger messageCounter = new AtomicInteger(0);
    private static String appInstanceId = ZonedDateTime.now().format(DateTimeFormatter.ofPattern("HHmmss"));

    static class MessageHandler implements MessageReceiver {

        @Override
        public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
            try {
                TimeUnit.SECONDS.sleep(10);
                System.out.println(
                        ZonedDateTime.now().format(DateTimeFormatter.ofPattern("HH:mm:ss")) + ",\t"
                        + "App instance id" + appInstanceId
                        + ",\tProcessing id: " + messageCounter.incrementAndGet()
                        + ",\tmessage content:" + message.getData().toStringUtf8()
                );
                consumer.ack();
            }
            catch (InterruptedException e){
                consumer.nack();
            }
        }
    }

    /** Receive messages over a subscription. */
    public static void main(String... args) throws Exception {
        // set subscriber id, eg. my-sub
        String projectId = args[0];
        String subscriptionId = args[1];
        ProjectSubscriptionName subscriptionName = ProjectSubscriptionName.of( projectId, subscriptionId);
        Subscriber subscriber = null;
        try {
            // we create a single threaded subscriber with the most restrictive flow control setting:
            subscriber =
                    Subscriber.newBuilder(subscriptionName, new MessageHandler())
                            .setFlowControlSettings(FlowControlSettings.newBuilder().setMaxOutstandingElementCount(1L).build())
                            .setExecutorProvider(InstantiatingExecutorProvider.newBuilder().setExecutorThreadCount(1).build())
                            .build();
            subscriber.startAsync().awaitTerminated();
        } finally {
            if (subscriber != null) {
                subscriber.stopAsync();
            }
        }
    }
}

    <dependency>
        <groupId>com.google.cloud</groupId>
        <artifactId>google-cloud-pubsub</artifactId>
        <version>0.42.1-beta</version>
    </dependency>

Logs
# start first client at 17:00:34
17:00:47,	App instance id170034,	Processing id: 1,	message content:1
17:00:57,	App instance id170034,	Processing id: 2,	message content:5
17:01:07,	App instance id170034,	Processing id: 3,	message content:9
17:01:17,	App instance id170034,	Processing id: 4,	message content:4
17:01:27,	App instance id170034,	Processing id: 5,	message content:8
17:01:37,	App instance id170034,	Processing id: 6,	message content:3
17:01:47,	App instance id170034,	Processing id: 7,	message content:7
17:01:57,	App instance id170034,	Processing id: 8,	message content:2
17:02:07,	App instance id170034,	Processing id: 9,	message content:6
17:02:17,	App instance id170034,	Processing id: 10,	message content:13
17:02:27,	App instance id170034,	Processing id: 11,	message content:17
17:02:37,	App instance id170034,	Processing id: 12,	message content:12
17:02:47,	App instance id170034,	Processing id: 13,	message content:16
17:02:57,	App instance id170034,	Processing id: 14,	message content:11
17:03:07,	App instance id170034,	Processing id: 15,	message content:15
17:03:17,	App instance id170034,	Processing id: 16,	message content:10
17:03:27,	App instance id170034,	Processing id: 17,	message content:14
17:03:37,	App instance id170034,	Processing id: 18,	message content:22

# start second and third clients around here: they generate no logs

17:03:47,	App instance id170034,	Processing id: 19,	message content:26
17:03:57,	App instance id170034,	Processing id: 20,	message content:21
17:04:07,	App instance id170034,	Processing id: 21,	message content:25
17:04:17,	App instance id170034,	Processing id: 22,	message content:19
# kill first client

Process finished with exit code 130 (interrupted by signal 2: SIGINT)
# second client logs (note they start "10 seconds" after the first client is killed
17:04:35,	App instance id170328,	Processing id: 1,	message content:39
17:04:45,	App instance id170328,	Processing id: 2,	message content:47
17:04:55,	App instance id170328,	Processing id: 3,	message content:20
17:05:05,	App instance id170328,	Processing id: 4,	message content:40
17:05:15,	App instance id170328,	Processing id: 5,	message content:38
17:05:25,	App instance id170328,	Processing id: 6,	message content:46
17:05:35,	App instance id170328,	Processing id: 7,	message content:27
17:05:45,	App instance id170328,	Processing id: 8,	message content:42
17:05:55,	App instance id170328,	Processing id: 9,	message content:55
17:06:05,	App instance id170328,	Processing id: 10,	message content:48
17:06:15,	App instance id170328,	Processing id: 11,	message content:56
17:06:25,	App instance id170328,	Processing id: 12,	message content:54
17:06:35,	App instance id170328,	Processing id: 13,	message content:49
17:06:45,	App instance id170328,	Processing id: 14,	message content:57
17:07:39,	App instance id170328,	Processing id: 15,	message content:18
17:07:49,	App instance id170328,	Processing id: 16,	message content:23
17:08:19,	App instance id170328,	Processing id: 17,	message content:30
17:08:29,	App instance id170328,	Processing id: 18,	message content:34
17:09:39,	App instance id170328,	Processing id: 19,	message content:32

# third client logs (note they start "10 seconds" after the first client is killed
17:04:35,	App instance id170328,	Processing id: 1,	message content:39
17:04:45,	App instance id170328,	Processing id: 2,	message content:47
17:04:55,	App instance id170328,	Processing id: 3,	message content:20
17:05:05,	App instance id170328,	Processing id: 4,	message content:40
17:05:15,	App instance id170328,	Processing id: 5,	message content:38
17:05:25,	App instance id170328,	Processing id: 6,	message content:46
17:05:35,	App instance id170328,	Processing id: 7,	message content:27
17:05:45,	App instance id170328,	Processing id: 8,	message content:42
17:05:55,	App instance id170328,	Processing id: 9,	message content:55
17:06:05,	App instance id170328,	Processing id: 10,	message content:48
17:06:15,	App instance id170328,	Processing id: 11,	message content:56
17:06:25,	App instance id170328,	Processing id: 12,	message content:54
17:06:35,	App instance id170328,	Processing id: 13,	message content:49
17:06:45,	App instance id170328,	Processing id: 14,	message content:57
17:07:39,	App instance id170328,	Processing id: 15,	message content:18
17:07:49,	App instance id170328,	Processing id: 16,	message content:23
17:08:19,	App instance id170328,	Processing id: 17,	message content:30
17:08:29,	App instance id170328,	Processing id: 18,	message content:34
17:09:39,	App instance id170328,	Processing id: 19,	message content:32

kir-titievsky · 2018-04-13T21:19:45Z

Max D has suggested a simple fix for this: close the stream whenever the buffer is full. This would not have worked when we used the same streamingPull connection for acks as for pulls. But using a separate connection for acks makes this feasible.

The cost is additional bandwidth, but at least the "stuckness" and load-balancing behavior is sane.

kir-titievsky · 2018-04-13T21:21:43Z

@saturnism FYI -- and thanks for discovering this.

pongad · 2018-04-18T18:12:26Z

This should have been alleviated by googleapis/google-cloud-java#3147 since we'd only open 1 connection by default and fewer messages should get stuck in buffer.

kir-titievsky · 2018-04-18T19:29:24Z

Not quite enough, since the buffer per 1 connection is still 10MB. So in my case of < 1000 messages, the behavior still looks ... surprising.

…

On Wed, Apr 18, 2018 at 2:12 PM Michael Darakananda < ***@***.***> wrote: This should have been alleviated by googleapis/google-cloud-java#3147 <googleapis/google-cloud-java#3147> since we'd only open 1 connection by default and fewer messages should get stuck in buffer. — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <https://github.com/GoogleCloudPlatform/google-cloud-java/issues/3152#issuecomment-382479317>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ARrMFvh8HRkKD_6vjC14LZrd0mnzpCpIks5tp4INgaJpZM4TUbbt> .

-- Kir Titievsky | Product Manager | Google Cloud Pub/Sub <https://cloud.google.com/pubsub/overview>

willy2706 · 2018-04-19T17:14:49Z

@pongad I agree with @kir-titievsky. This issue occurs in my production when there are lots of messages published to a topic.

At last, I downgraded to v0.39.0-beta and used PollingSubscriberConnection

justbaum30 · 2018-09-20T16:15:57Z

I'm also running into this issue on my own project. Almost same exact criteria as @kir-titievsky laid out above where we only want each instance of our application to be picking up 1 message at a time because the processing time of the message is relatively long, yet variable. We're seeing situations where a single subscriber hoards messages while another is left with nothing to do even though the max element count is set to 1 with a single executor.

Are there any updates on this issue? It's looking like we're going to have to change to synchronous polling instead because of this issue. Thanks for your time.

meltsufin · 2018-09-20T19:38:39Z

@kir-titievsky This sounds like a serious bug, not a feature request. Or, am I missing something?

kir-titievsky · 2018-09-20T19:53:15Z

The "hoarding" is a known limitation of streamingPull (see https://cloud.google.com/pubsub/docs/pull -- at the very end there), which is relevant when you have a large backlog of small tasks that take >>100ms to process. The workaround is to use synchronous pull, documented in the same place.

…

On Thu, Sep 20, 2018 at 3:38 PM Mike Eltsufin ***@***.***> wrote: @kir-titievsky <https://github.com/kir-titievsky> This sounds like a serious bug, not a feature request. Or, am I missing something? — You are receiving this because you were mentioned. Reply to this email directly, view it on GitHub <https://github.com/GoogleCloudPlatform/google-cloud-java/issues/3152#issuecomment-423307245>, or mute the thread <https://github.com/notifications/unsubscribe-auth/ARrMFo1giv9H2jvSM2UO38oYrmDo_6ORks5uc-7JgaJpZM4TUbbt> .

-- Kir Titievsky | Product Manager | Google Cloud Pub/Sub <https://cloud.google.com/pubsub/overview>

roku6185 · 2018-12-12T13:59:13Z

Are there any code snippets to avoid "hoarding" using sync. pull? I'm having something similar to this issue on some jobs. Do I have to take care of extending the ack deadline myself when using sync. pull?

meltsufin · 2018-12-12T14:49:59Z

@roku6185 The discussion about "hoarding" was in regards to the streaming (async) pull. Can you explain your issue with the sync pull? In sync pull you can specify the limit on how many messages to retrieve at a time. So, hoarding should not be an issue, in theory.

roku6185 · 2018-12-12T17:57:19Z

My bad, let my rephrase. I'm using async. pull today and I'm having some kind of problem with retries. I think this has to do with the ack deadlnes not being extended properly. So I'm looking for some code snippet on how to implement the sync. pull in order to avoid this problem. I don't know if my problem is a "hoarding" problem, but my setup is similar to what @justbaum30 described above.

meltsufin · 2018-12-12T18:23:18Z

Problems with streaming pull do exist as laid out in the above discussion.
We have a task to completely move away from streaming pull and use synchronous pull instead.
You can use synchronous pull through the Pub/Sub template.
If you think there is a specific issue we don't know about in the streaming pull, we will probably need an example that reproduces it to debug further.

ajaaym · 2019-03-08T19:07:27Z

@meltsufin @roku6185 @kir-titievsky You can use below settings to avoid getting lot of message buffered in to client.

    Subscriber subscriber = Subscriber.newBuilder("", new MessageReceiver() {
      @Override public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
        consumer.ack();
      }
    }).setChannelProvider(SubscriberStubSettings.defaultGrpcTransportProviderBuilder().setChannelConfigurator(
            new ApiFunction<ManagedChannelBuilder, ManagedChannelBuilder>() {
              @Override public ManagedChannelBuilder apply(ManagedChannelBuilder input) {
                ((NettyChannelBuilder)input).flowControlWindow(4096); // try to keep around 10 * avg msg size
                return input;
              }
            }).build())
        .setParallelPullCount(1) //try to keep it to lower number if you dont want to buffer more message, 1 here will buffer 10 msg when flowcontrolwindow is configured properly
.setFlowControlSettings(FlowControlSettings.newBuilder().setMaxOutstandingElementCount(10).build()) //make here num puller * 10
        .build();

meltsufin · 2019-03-08T19:35:12Z

@ajaaym This pretty ugly, but may be a valuable workaround for some people. I'd prefer to see this exposed a bit more conveniently in the Pub/Sub client library.
@kir-titievsky Is this an approach you recommend?

ajaaym · 2019-03-08T19:37:06Z

@meltsufin yes it is, but can be used until proper solution is out.

sduskis · 2019-05-03T15:55:27Z

@ajaaym and I discussed the possibility of having a new maxMessageSize variable that would allow the client to properly size the flowControlWindow

meredithslota · 2021-01-12T18:47:13Z

@kir-titievsky and @kamalaboulhosn — is this feature request still relevant? Feels like it might be made obsolete by the recent discussions re: flow control.

kir-titievsky · 2021-01-12T19:57:04Z

yes, this should now be solved by the flow control api changes.

meredithslota · 2021-02-05T18:19:28Z

Ok, great. I'm just going to close this FR then - let me know if I've misunderstood. :)

dgrudenic-qumu · 2021-03-24T14:33:24Z

So anyone knows if this feature is available so that example above works as expected?

meredithslota · 2021-03-24T17:51:51Z

https://cloud.google.com/pubsub/docs/pull with the included sample https://cloud.google.com/pubsub/docs/samples/pubsub-subscriber-flow-settings#pubsub_subscriber_flow_settings-java shows using flow control with a Pub/Sub subscriber. I'm not sure which portion of the thread you mean re: works as expected, but hopefully that'll get you started!

dgrudenic-qumu · 2021-03-24T18:14:17Z

We actually experienced same behavior mentioned in example above.

So if you have 3 subscribers on same subscription/topic and you want that each subscriber is processing 1 message at the time which is kind of standard point to point workflow it is not really possible with FlowControlSettings and setting setMaxOutstandingElementCount=1 because as soon as first subscriber gets message it will buffer more than 1 message.

Let me know if this is not the case and example above is actually working.

meredithslota · 2021-03-25T16:02:33Z

I opened a new issue to debug the issue you are having, since the original issue was a feature request.

chingor13 transferred this issue from googleapis/google-cloud-java Dec 4, 2019

chingor13 added the type: feature request ‘Nice-to-have’ improvement, new feature or different behavior or design. label Dec 4, 2019

google-cloud-label-sync bot added the api: pubsub Issues related to the googleapis/java-pubsub API. label Jan 29, 2020

meredithslota closed this as completed Feb 5, 2021

meredithslota self-assigned this Feb 5, 2021

meredithslota mentioned this issue Mar 25, 2021

Issue using FlowControlSettings and setting setMaxOutstandingElementCount = 1 #582

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Pub/Sub: messages stuck in buffer, preventing proper load balancing #17

Pub/Sub: messages stuck in buffer, preventing proper load balancing #17

kir-titievsky commented Apr 13, 2018

kir-titievsky commented Apr 13, 2018

kir-titievsky commented Apr 13, 2018

pongad commented Apr 18, 2018

kir-titievsky commented Apr 18, 2018 via email

willy2706 commented Apr 19, 2018 •

edited

Loading

justbaum30 commented Sep 20, 2018

meltsufin commented Sep 20, 2018

kir-titievsky commented Sep 20, 2018 via email

roku6185 commented Dec 12, 2018 •

edited

Loading

meltsufin commented Dec 12, 2018

roku6185 commented Dec 12, 2018

meltsufin commented Dec 12, 2018

ajaaym commented Mar 8, 2019 •

edited

Loading

meltsufin commented Mar 8, 2019

ajaaym commented Mar 8, 2019

sduskis commented May 3, 2019

meredithslota commented Jan 12, 2021

kir-titievsky commented Jan 12, 2021

meredithslota commented Feb 5, 2021

dgrudenic-qumu commented Mar 24, 2021

meredithslota commented Mar 24, 2021

dgrudenic-qumu commented Mar 24, 2021

meredithslota commented Mar 25, 2021

Pub/Sub: messages stuck in buffer, preventing proper load balancing #17

Pub/Sub: messages stuck in buffer, preventing proper load balancing #17

Comments

kir-titievsky commented Apr 13, 2018

kir-titievsky commented Apr 13, 2018

kir-titievsky commented Apr 13, 2018

pongad commented Apr 18, 2018

kir-titievsky commented Apr 18, 2018 via email

willy2706 commented Apr 19, 2018 • edited Loading

justbaum30 commented Sep 20, 2018

meltsufin commented Sep 20, 2018

kir-titievsky commented Sep 20, 2018 via email

roku6185 commented Dec 12, 2018 • edited Loading

meltsufin commented Dec 12, 2018

roku6185 commented Dec 12, 2018

meltsufin commented Dec 12, 2018

ajaaym commented Mar 8, 2019 • edited Loading

meltsufin commented Mar 8, 2019

ajaaym commented Mar 8, 2019

sduskis commented May 3, 2019

meredithslota commented Jan 12, 2021

kir-titievsky commented Jan 12, 2021

meredithslota commented Feb 5, 2021

dgrudenic-qumu commented Mar 24, 2021

meredithslota commented Mar 24, 2021

dgrudenic-qumu commented Mar 24, 2021

meredithslota commented Mar 25, 2021

willy2706 commented Apr 19, 2018 •

edited

Loading

roku6185 commented Dec 12, 2018 •

edited

Loading

ajaaym commented Mar 8, 2019 •

edited

Loading