Merge pull request #1838 from Logflare/feat/pipeline-tweaks

feat: tweaks broadway pipeline batch size, batch timeout
Logflare · Nov 22, 2023 · 6fd0845 · 6fd0845
2 parents 323f869 + 4472573
commit 6fd0845
Show file tree

Hide file tree

Showing 2 changed files with 5 additions and 3 deletions.
diff --git a/lib/logflare/source/bigquery/pipeline.ex b/lib/logflare/source/bigquery/pipeline.ex
@@ -27,15 +27,16 @@ defmodule Logflare.Source.BigQuery.Pipeline do
 
     Broadway.start_link(__MODULE__,
       name: name(source.token),
+      max_restarts: 10,
       producer: [
         module: {BufferProducer, rls},
         hibernate_after: 30_000
       ],
       processors: [
-        default: [concurrency: 1]
+        default: [concurrency: max_batchers]
       ],
       batchers: [
-        bq: [concurrency: max_batchers, batch_size: 250, batch_timeout: 1000]
+        bq: [concurrency: max_batchers, batch_size: 500, batch_timeout: 1_500]
       ],
       context: rls
     )

diff --git a/test/logflare/logs/logs_test.exs b/test/logflare/logs/logs_test.exs
@@ -124,7 +124,8 @@ defmodule Logflare.LogsTest do
       ]
 
       assert :ok = Logs.ingest_logs(batch, source)
-      :timer.sleep(1_500)
+      # batcher timneout is 1_500
+      :timer.sleep(2_000)
     end
   end