Poll heartbeat rate #250

paualarco · 2021-01-24T18:18:02Z

The main motivations for this PR is to tackle two existing issues #94 and #101.
Rebasing existing pending PR: #104, so credits for the author, @voidconductor 🙏

Proper commitAsync implementation that removed callback from methods arguments itself, now completed in this callback, and it was also never used actually used in previous implementations. No api changes.
Polling strategy for keeping the consumer alive and not trigger partition reassignment in slow subscribers, which avoids doing the hack of decreasing poll.max.records. Consumer can be rebalanced if poll is not called before max.poll.interval.ms #101.
The polling rate is only configurable internally for performance analysis on benchmarks repos. Ideally I we would just
This polling is achieved by performing poll(0) from a task running concurrently in the background, which in case of error it the schedule will report it, without stopping the subscription. Although it should never happen since the heartbeat poll is only called if the consumer is paused, and both processes are call synchronized.

Add consumer polling (monix#94) AutoCommit polling logic (monix#94) Remove nested task loop Remove redundant atomic refs Add test for (monix#101) Apply changes to older versions

Avasil

Stellar work, we're very close to finally fixing this issue! :D

Avasil · 2021-02-11T19:10:29Z

kafka-0.10.x/src/main/scala/monix/kafka/KafkaConsumerObservable.scala

-      }
-      feedTask.runAsync(cb)
+      val startConsuming =
+        Resource


We immediately use the Resource so I feel like bracket would be a better fit

kafka-0.10.x/src/main/scala/monix/kafka/KafkaConsumerObservable.scala

Avasil · 2021-02-11T19:39:45Z

kafka-0.10.x/src/main/resources/monix/kafka/default.conf

@@ -77,4 +77,6 @@ kafka {
  monix.observable.commit.type = "sync"
  # Possible values: before-ack, after-ack or no-ack
  monix.observable.commit.order = "after-ack"
+  # Internal kafka heartbeat that avoids partition reasignment when the upstream subscribers is slow.
+  monix.observable.poll.heartbeat.rate.ms = 100


Is there any recommended ratio between poll heartbeats and max.poll.interval.ms?

@Avasil in order to respond to that question, I have created benchmark with different poll intervals:

Benchmark Mode Cnt Score Error Units ConsumerBenchmark.monix_auto_commit10ms thrpt 10 17.266 ± 2.231 ops/s ConsumerBenchmark.monix_manual_commit_heartbeat1000ms thrpt 10 2.971 ± 0.009 ops/s ConsumerBenchmark.monix_manual_commit_heartbeat100ms thrpt 10 9.477 ± 0.064 ops/s ConsumerBenchmark.monix_manual_commit_heartbeat10ms thrpt 10 14.710 ± 1.660 ops/s ConsumerBenchmark.monix_manual_commit_heartbeat1ms thrpt 10 15.494 ± 4.163 ops/s

Not sure why the performance is degraded with the increment of the pollHeartBeat, from the numbers, I would say that 10.ms would be the best fit. I have made the property configurable only internally, since I think the user does not need to know about it. Any advice?

I would expect that we could do this heartbeat very rarely.
Is it only for manual commit?

I think the commitAsync can only complete after poll so that could be a potential cause? Perhaps we could do a heartbeat on commit as well? Akka Streams seems to be doing something like that but I have only had a quick glance and could be completely wrong.

I have made the property configurable only internally, since I think the user does not need to know about it.

Is this the current state?

Yes I did that after seeing that altering this value could impact substantially performance and the user should not be aware of it.
Ley me try your suggestion of polling after commit!

kafka-0.10.x/src/test/scala/monix/kafka/MonixKafkaTopicListTest.scala

Avasil · 2021-02-11T19:48:04Z

kafka-0.10.x/src/test/scala/monix/kafka/MonixKafkaTopicListTest.scala

+    }
+  }
+
+  test("slow batches autocommit processing doesn't cause rebalancing") {


What if there is a rebalancing caused by other reasons? For instance, if another consumer shuts down.
Can we test if we're protected against such scenarios?

I'm thinking that maybe some tests with multiple consumers could be useful. Or if there are some tools to force the repartitioning

paualarco · 2021-02-11T20:30:27Z

@Avasil thank you for all the comments! Will address them and do review again 🧐

Block Fix Binded connection First benchmark for kafka producer Added kafka benchmark strategy plan Added sink and consumer benchmarks Producer results Akka Removed references to akka a Final

Revmoves travis-ci

paualarco · 2021-04-02T09:15:07Z

@Avasil first of all sorry for the amount of commits, it might make difficult to track the changes review, I assume they will be squashed at the end...

I have written specific new cases in PollHeartbeatTest for the different kafka versions (excluding 0.9.x), and benchmarks covering manual and auto-commit scenarios with different heartbeat interval values.
As part of that I ran the same benchmarks from latest release RC7, to have a base to compare with, and from there I've got two conclusions:

The higher pollHeartBeatInterval is, the slower the consumption becomes (although I am unsure what this relation is due to). So I ended up setting by default a heartbeat interval of 15ms.
On the other hand, I have noticed that the auto-commit-async has degraded and I am not sure why (I'm investigating it).

Finally, I was also tempted to move from travis to github actions and splitting the tests by the different kafka versions.

Avasil · 2021-04-02T11:12:50Z

first of all sorry for the amount of commits, it might make difficult to track the changes review, I assume they will be squashed at the end...

No worries, just let me know when you think it's good to re-review - thanks for the hard work!

As part of that I ran the same benchmarks from latest release RC7, to have a base to compare with, and from there I've got two conclusions:

I'll look at the benchmarks later but I feel like they can be unreliable if everything is hosted on one machine so it's hard to draw any conclusions

Finally, I was also tempted to move from travis to github actions and splitting the tests by the different kafka versions.

👍 Yeah, it needs to be done and I could setup sbt ci-release

paualarco · 2021-04-10T21:10:05Z

@Avasil I think it could already be re-reviewed :)

Avasil

I didn't look too much into benchmarks but the code seems good to me!

Would be nice to give it a try in a real project though.

@allantl do you happen to still use monix-kafka and would be able to try these changes if we release a snapshot ?

Avasil · 2021-04-11T14:11:11Z

CHANGES.md

@@ -1,3 +1,22 @@
+
+## Version 1.0.0-RC6 (January 23, 2021)


Seems like this one should be RC7 and Version 1.0.0-RC7 (April 18, 2020) below should be removed

Corrected, sorry I only intended to add the entry for RC7 that was missing.

Avasil · 2021-04-11T14:13:56Z

benchmarks/results/producer.md

+
+## RC7
+
+### 10iterations 1fork 1thread


Shouldn't we compare results of the same fork / threads setup?

Yep they should, will address that, although I think I just added them more as a base for future api changes since the producer logic was untouched.

Avasil · 2021-04-11T14:31:49Z

kafka-0.10.x/src/main/scala/monix/kafka/KafkaConsumerObservable.scala

+          Task.race(runLoop(c, out), pollHeartbeat(c).loopForever).void
+        } { consumer =>
+          // Forced asynchronous boundary
+          Task.evalAsync(consumer.synchronized(blocking(consumer.close()))).memoizeOnSuccess


memoize seems to be redundant, is there any situation where the finalizer would be called more than once?

paualarco · 2021-04-11T17:09:16Z

Would be nice to give it a try in a real project though.

That would be ideal, alternatively we may ask it in the main monix gitter channel too, and see if someone is interested in testing it.

paualarco added 10 commits January 17, 2021 12:43

Async commit completed with callback (monix#94)

0ad10cb

Add consumer polling (monix#94) AutoCommit polling logic (monix#94) Remove nested task loop Remove redundant atomic refs Add test for (monix#101) Apply changes to older versions

Tests passing

d56095d

Observable poll heartbeat rate

bc18d91

Resolves merge conflicts

fa380f6

Unused imports

7232568

Multiple threads test failing

32d238c

Test small poll heartbeat and slow upstream

622b180

ScalafmtAll

1365c7c

Removes lambda anonymous class for scala 11 compatibility

c6c1c32

Scaladoc fix

512f56c

Avasil reviewed Feb 11, 2021

View reviewed changes

paualarco added 18 commits March 21, 2021 10:16

MR Feedback

2651712

Unused import

336a9f2

Adding poll heartbeat tests

b835310

l

e612d64

Improved rebalancing test scenarios

ca5e98a

a

4bcf70c

Improved test

f93ae9a

Monix Kafka benchmarks for consumer, single and sink producer

1c8c5c1

Block Fix Binded connection First benchmark for kafka producer Added kafka benchmark strategy plan Added sink and consumer benchmarks Producer results Akka Removed references to akka a Final

Benchmarks

0bf0bff

Fs2

4974c62

Heartbeat poll

f13d855

Benchmarks updated

d3a4425

Private poll heartbeat rate

0dee12b

Some improvements

bc6eb42

Fix unexpected consumed records

eca6927

Benchmarks with different pollHeartbeatIntervals

6165418

New benchmark results

610eb74

More benchmarking

f0861e7

paualarco and others added 5 commits March 30, 2021 22:52

Merge branch 'master' into poll-heartbeat-rate

ff7f422

Update changelog and trigger pipeline

6d1ebbf

Test indentation fix

38f7d47

Add logback in test

bac3175

Add unchanged topic partition test case

444c2a2

paualarco marked this pull request as draft April 1, 2021 22:31

paualarco added 6 commits April 2, 2021 00:34

Kafka tests running on github actions

b0316b6

Renames github workflows folder

38b31c7

Updates scala action

c61857f

Removes tests for kafka 0.9.x

edeb179

Revmoves travis-ci

Add benchmark with autocomit async and sync scenarios

8c09b6a

Removed benchmark results file

d54ed10

paualarco added 7 commits April 3, 2021 10:01

Some clean up

9bcd8ac

Use pollTimeout and updated benchmark results

6f2fa50

ScalafmtAll

50c3d9d

Ignore Ds_store file

f7991ca

Removes logback file from main

1c71a68

Remove unnecessary stuff

f05abe6

Remove unused import

12e75a8

paualarco marked this pull request as ready for review April 3, 2021 14:16

paualarco added 2 commits April 9, 2021 20:35

Benchmarks readme

10cbf9f

Small correction

32c5bd7

Avasil reviewed Apr 11, 2021

View reviewed changes

MR Feedback

1526702

Avasil mentioned this pull request Nov 28, 2021

Implements a KafkaConsumerResource to solve consumer already closed i… #306

Open

Avasil merged commit fd89177 into monix:master Mar 24, 2022

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Poll heartbeat rate #250

Poll heartbeat rate #250

paualarco commented Jan 24, 2021 •

edited

Loading

Avasil left a comment

Avasil Feb 11, 2021

Avasil Feb 11, 2021

paualarco Mar 27, 2021

Avasil Apr 11, 2021

paualarco Apr 11, 2021

Avasil Feb 11, 2021

paualarco commented Feb 11, 2021

paualarco commented Apr 2, 2021

Avasil commented Apr 2, 2021

paualarco commented Apr 10, 2021

Avasil left a comment

Avasil Apr 11, 2021

paualarco Apr 11, 2021

Avasil Apr 11, 2021

paualarco Apr 11, 2021

Avasil Apr 11, 2021

paualarco Apr 11, 2021

paualarco commented Apr 11, 2021

Poll heartbeat rate #250

Poll heartbeat rate #250

Conversation

paualarco commented Jan 24, 2021 • edited Loading

Avasil left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

paualarco commented Feb 11, 2021

paualarco commented Apr 2, 2021

Avasil commented Apr 2, 2021

paualarco commented Apr 10, 2021

Avasil left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

paualarco commented Apr 11, 2021

paualarco commented Jan 24, 2021 •

edited

Loading