Throttle Kafka consumption to the file writing speed #686

danesss · 2023-04-03T15:38:21Z

Issue

ECDC-3253: We currently read from Kakfa at max speed and buffer into memory, which can cause out-of-memory issues if Kafka is faster than the GPFS filesystem.

Description of work

We adapt the consumption rate to the file writing rate:

We add a new configuration setting for the size of write queue --max-queued-writes
We add a method in MessageWriter to be able to check current queue size
We extend StreamController to
- Perform a periodic check of the write queue size
- Pause all consumers for 200ms if queue is larger than configured
Topic and Partition classes have new methods to support pause and resume of the consumers.
Unit tests are added to verify pause/resume behaviour at the Partition level.
- To test the "queue_size -> pause" logic further changes are needed so they will be explored in separate PR: Improve stream controller tests #689

Nominate for Group Code Review

Nominate for code review

Reminder

Changes should be documented in changes.md

We need to adapt the consumption rate to the file writing rate to avoid out-of-memory issues due to WriteQueue buffering.

…_speed

cow-bot · 2023-04-03T15:59:42Z

Code Coverage
(Lines Exec Cover)
TOTAL 5590 3771 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-03T15:59:58Z

Code Coverage
(Lines Exec Cover)
TOTAL 5590 3778 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-04T08:31:16Z

Code Coverage
(Lines Exec Cover)
TOTAL 5590 3774 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-04T08:47:15Z

Code Coverage
(Lines Exec Cover)
TOTAL 5590 3781 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-04T09:02:52Z

Code Coverage
(Lines Exec Cover)
TOTAL 5590 3774 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-04T09:42:05Z

Code Coverage
(Lines Exec Cover)
TOTAL 5590 3780 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-04T13:10:17Z

Code Coverage
(Lines Exec Cover)
TOTAL 5600 3781 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-04T13:19:03Z

Code Coverage
(Lines Exec Cover)
TOTAL 5603 3775 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-04T14:38:05Z

Code Coverage
(Lines Exec Cover)
TOTAL 5603 3775 67%
For more detail see Cobertura report in Jenkins interface

This reverts commit 29e8fff.

This reverts commit d12ffa5.

cow-bot · 2023-04-04T14:59:00Z

Code Coverage
(Lines Exec Cover)
TOTAL 5603 3775 67%
For more detail see Cobertura report in Jenkins interface

…_speed

cow-bot · 2023-04-04T16:04:44Z

Code Coverage
(Lines Exec Cover)
TOTAL 5603 3775 67%
For more detail see Cobertura report in Jenkins interface

We need to adapt the consumption rate to the file writing rate to avoid out-of-memory issues due to WriteQueue buffering.

cow-bot · 2023-04-12T09:19:03Z

Code Coverage
(Lines Exec Cover)
TOTAL 5607 3774 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-15T03:33:07Z

Code Coverage
(Lines Exec Cover)
TOTAL 5606 3777 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-15T03:37:08Z

Code Coverage
(Lines Exec Cover)
TOTAL 5606 3773 67%
For more detail see Cobertura report in Jenkins interface

src/Stream/Partition.h

src/Stream/Topic.h

src/StreamController.cpp

mattclarke · 2023-04-19T11:31:46Z

src/StreamController.h

+  /// Consumers are stopped when the write queue is larger than
+  /// StreamerOptions.MaxQueuedWrites. This variable defines the ratio below
+  /// which the consumers will be resumed.
+  float const QueuedWritesResumeThreshold{0.8F};


src/StreamController.h

mattclarke

Looks good.
I think the names changes and names for new items are a big improvement 👍

cow-bot · 2023-04-19T11:48:46Z

Code Coverage
(Lines Exec Cover)
TOTAL 5606 3774 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-19T11:57:27Z

Code Coverage
(Lines Exec Cover)
TOTAL 5606 3773 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-19T12:04:40Z

Code Coverage
(Lines Exec Cover)
TOTAL 5606 3773 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-19T12:12:41Z

Code Coverage
(Lines Exec Cover)
TOTAL 5606 3774 67%
For more detail see Cobertura report in Jenkins interface

StreamersPaused atomic bool was set externally when throttling consumers, instead of within the pauseStreamers() method. It seems reasonable to ensure the mutex is set by making the setter part of the method.

…cope StreamersPaused atomic bool was set externally when throttling consumers, instead of within the pauseStreamers() method. It seems reasonable to ensure the mutex is set by making the setter part of the method.

cow-bot · 2023-04-19T14:29:21Z

Code Coverage
(Lines Exec Cover)
TOTAL 5606 3774 67%
For more detail see Cobertura report in Jenkins interface

cow-bot · 2023-04-19T14:41:41Z

Code Coverage
(Lines Exec Cover)
TOTAL 5606 3774 67%
For more detail see Cobertura report in Jenkins interface

danesss and others added 2 commits April 3, 2023 16:56

Feat(stream): Add support to pause consumers

e7a800f

We need to adapt the consumption rate to the file writing rate to avoid out-of-memory issues due to WriteQueue buffering.

Merge branch 'main' into ECDC-3253_throttle_consumers_to_file_writing…

7450354

…_speed

CI: Make Jenkins show cppcheck errors

d12ffa5

danesss added 2 commits April 4, 2023 10:33

CI: Fix Jenkins syntax error

29e8fff

Refact: Make some functions const

ec4a94a

danesss added 2 commits April 4, 2023 11:22

Refact(stream): Make pauseConcusmers functions public

1023ec4

Doc(partition): Update docstrings

0a7ef10

danesss added 4 commits April 4, 2023 14:21

Refact(StreamController): Support several periodic checks

0c4cfe3

Refact(StreamController): Rename StreamerOptions variable

374ee79

Feat(stream): Pause consumers if write queue is full

81275a9

Feat(stream): Add StreamersPaused flag in StreamController

2b724cf

danesss added 2 commits April 4, 2023 16:45

Revert "CI: Fix Jenkins syntax error"

bbd9e5a

This reverts commit 29e8fff.

Revert "CI: Make Jenkins show cppcheck errors"

976cddf

This reverts commit d12ffa5.

Merge branch 'main' into ECDC-3253_throttle_consumers_to_file_writing…

825b0e2

…_speed

danesss added 5 commits April 4, 2023 18:12

Feat(stream): Add support to pause consumers

8aac507

We need to adapt the consumption rate to the file writing rate to avoid out-of-memory issues due to WriteQueue buffering.

CI: Make Jenkins show cppcheck errors

7e93043

CI: Fix Jenkins syntax error

7b5f597

Refact: Make some functions const

ac5d57e

Refact(stream): Make pauseConcusmers functions public

9cd5cd9

Merge commit '2d2694cd663f545067cf9f9a778a53d40ebcc43a' into HEAD

33480a8