google pubsub to google bq not working

[mittal-jatin]

Hi, I am using benthos to stream data from google pubsub to google bigQuery using the below config.yaml :

input:
  label: my_input_label
  gcp_pubsub:
    project: "my_project_name"
    subscription: "my_subscription_name"
    sync: false
pipeline:
  processors:
    - log:
        level: INFO
        message: Hello World
output:
  label: my_output_label
  gcp_bigquery:
    project: "my_project_name"
    dataset: "dataset_name"
    table: "table_name"
    format: NEWLINE_DELIMITED_JSON
    max_in_flight: 10000
    write_disposition: WRITE_APPEND
    batching:
      count: 10000
      period: 10s

In documentation, I could not differentiate between max_in_flight and batching.count clearly. Can someone give more context on these two fields. Also, when I tried changing max_in_flight value to 10000 and batching.count to 1000, I was getting some messages in BQ but not all.
For the above configurations, I can see the Hello World logs in terminal but on pubsub console 9000 messages are unacknowledged and there is only 1000 records in BQ.
Can someone tell where am I missing ?
Using benthos version - 4.0.0

[Jeffail]

Hey @mittal-jatin the docs are a bit misleading for the max_in_flight field, it's actually the number of batches that could be in flight at a given time. In your case I would keep it fairly low (maybe tune it down to 10) as right now you're probably flooding GCP which is causing all the unacked messages as they're consumed but pending acknowledgement until the backlog is processed.

[mittal-jatin]

so if max_in_flight is 10000 and batching.count is 10000, how many batches would it be sending to BQ in one go ? And what will be the batch size?
and, if max_in_flight is 10 and batching.count is 10000, how many batches will it be sending to BQ in one go ? And what will be the batch size in that case?

[mittal-jatin]

Also, I tried with max_in_flight as 10 and batching.count as 10000, after inserting some data, it starts throwing error -

msg="Failed to send message to gcp_bigquery: failed to complete bigquery job successfully: {Location: \"load_job_per_partitioned_table.long\"; Message: \"Quota exceeded: Your table exceeded quota for Imports/appends/dml per partitioned table.

[mihaitodor]

In case someone else bumps into this, as mentioned in a longer thread on the Golang Slack, you might want to start with max_in_flight: 1 and make sure your batches don't end up (a lot) smaller than the configured batch.count (you might want to increase this gradually, to optmise the output network throughput), which can happen if your input isn't producing enough messages during the configured batching.period. You can use a log processor under batching.processors to print the size of each emitted batch with by setting message: ${! batch_size() }. Once you get this to work, you can try to increase max_in_flight to 2 then 3 then 4 etc until you don't see any more performance benefit. You'll have to keep an eye on the batch size during each increase and, also, make sure that Benthos doesn't end up using more memory than it has available. You can monitor that using the exposed metrics by using the prometheus exporter with add_go_metrics: true.