Merge pull request #21 from NoteGio/feature/configurable-flush

ethdb/cdc: Make kafka flush frequency configurable
NoteGio · Oct 10, 2020 · 6b05247 · 6b05247
2 parents 3448391 + 157b150
commit 6b05247
Showing 1 changed file with 4 additions and 0 deletions.
diff --git a/ethdb/cdc/kafka.go b/ethdb/cdc/kafka.go
@@ -57,6 +57,10 @@ func ParseKafkaURL(brokerURL string) ([]string, *sarama.Config) {
     config.Producer.Retry.Max = 10000000
   }
 
+  if val, err := strconv.Atoi(parsedURL.Query().Get("queue.buffering.max.ms")); err == nil {
+    config.Producer.Flush.Frequency = time.Duration(val) * time.Millisecond
+  }
+
   if val, err := strconv.Atoi(parsedURL.Query().Get("retry.backoff.ms")); err == nil {
     config.Producer.Retry.Backoff = time.Duration(val) * time.Millisecond
   }