Asynchronous Subscription Begins to Fail After Some Time

[tristeybro]

Hello! I have a long-running script that is continuously pulling messages from a pubsublite topic asynchronously and executing them as they come in. This works fine for a while, but eventually runs into an AioRpcError due to receiving a FlowControlRequest that that tried to remove messages. Once that happens, the subscription continues to fail indefinitely. See below.

Environment details

• OS type and version: Linux
• Python version: 3.9.7
• pip version: 21.2.4
• google-cloud-pubsublite version: 1.2.0

Code example

import asyncio
from google.cloud.pubsublite.cloudpubsub import AsyncSubscriberClient
from google.cloud.pubsublite.types import (
    CloudRegion,
    CloudZone,
    FlowControlSettings,
    SubscriptionPath,
)
from google.cloud.pubsub_v1.types import PubsubMessage

per_partition_flow_control_settings = FlowControlSettings(
    messages_outstanding=20,
    # 10 MiB. Must be greater than the allowed size of the largest message (1 MiB).
    bytes_outstanding=10 * 1024 * 1024,
)

async def process_message():
    ...


async def main():
  async with AsyncSubscriberClient() as subscriber_client:
      streaming_pull_future = await subscriber_client.subscribe(
          subscription_path,
          per_partition_flow_control_settings=per_partition_flow_control_settings,
      )
      async for message in streaming_pull_future:
          process_message_coro = process_message(message)
          asyncio.create_task(process_message_coro)

if __name__ == "__main__":
    asyncio.run(main())

Stack trace

Traceback (most recent call last):
  File "/usr/local/lib/python3.9/site-packages/google/api_core/grpc_helpers_async.py", line 105, in _wrapped_aiter
    async for response in self._call:  # pragma: no branch
  File "/usr/local/lib/python3.9/site-packages/grpc/aio/_call.py", line 326, in _fetch_stream_responses
  await self._raise_for_status()
  File "/usr/local/lib/python3.9/site-packages/grpc/aio/_call.py", line 236, in _raise_for_status
  raise _create_rpc_error(await self.initial_metadata(), await
  grpc.aio._call.AioRpcError: <AioRpcError of RPC that terminated with:
    status = StatusCode.INVALID_ARGUMENT
    details = "Received FlowControlRequest that tried to remove messages"
    debug_error_string = "{"created":"@1643639089.487575412","description":"Error received from peer ipv4:108.177.111.95:443","file":"src/core/lib/surface/call.cc","file_line":1074,"grpc_message":"Received FlowControlRequest that tried to remove messages","grpc_status":3}"

Rest of Stack Trace

[dpcollins-google]

Hello,

I'm sorry you're having this issue. My guess is that you are hitting an edge case where you are somehow making the flow control count go negative.

EDIT: The message count is negative, not the byte count. It has a value of -1.

I will also fix the client to not send an invalid request when this happens.

-Daniel