Fix Partition Resumes #199

garyrussell · 2021-03-03T22:37:06Z

ConsumerEventLoop.PollEvent should not resume partitions that it, itself,
has not paused.

Prior to d908b81
pause/resume due to backpressure was protected by a boolean but it still unconditionally
resumed all partitions.

reintroduce the boolean and only resume if we paused
keep track of user-paused partitions and exclude them from resuming

Also fixes #190
Also fixes #198

garyrussell · 2021-03-03T22:38:42Z

cc/ @artembilan - I couldn't tag you for a review - maybe you aren't in this org?

artembilan · 2021-03-03T22:53:52Z

True. Here are people of this org: https://github.com/orgs/reactor/people.
But that still does not prevent us from reviewing.
Although looks like the merge is still only their prerogative 😄

simonbasle

need a few clarifications

simonbasle · 2021-03-04T08:45:09Z

src/main/java/reactor/kafka/receiver/internals/ConsumerEventLoop.java

+                            if (pausedByUs.getAndSet(false)) {
+                                Set<TopicPartition> toResume = new HashSet<>(consumer.assignment());
+                                toResume.removeAll(this.pausedByUser);
+                                consumer.resume(toResume);


is it relevant to keep the pausedByUser collection full after this point?
ie. all the branches where pausedByUs == false will clear the collection and refill it, but that clear is thus done in a lazy fashion

Good point; yes, it can be cleared here.

simonbasle · 2021-03-04T08:47:20Z

src/test/java/reactor/kafka/receiver/KafkaReceiverTest.java

@@ -1190,6 +1190,29 @@ public void abortTransaction() throws Exception {
        checkConsumedMessages(count, count * 3);
    }

+    @Test
+    public void userPause() throws InterruptedException {


does the test really cover externally paused topic partitions? ie. would there be a way to have 2 partitions paused and verify that the PollEvent distinguishes between the two more explicitly?

Yes; more test coverage is needed; I originally wrote this test just to demonstrate that the paused partition was incorrectly resumed.

The notion of user Vs. our pauses only came to me while fixing the bug, but I didn't enhance the test to cover that use case.

simonbasle · 2021-03-04T08:53:08Z

src/main/java/reactor/kafka/receiver/internals/ConsumerEventLoop.java

@@ -230,12 +236,22 @@ public void run() {
                    long r = requested;
                    if (r > 0) {
                        if (!awaitingTransaction.get()) {
-                            consumer.resume(consumer.assignment());
+                            if (pausedByUs.getAndSet(false)) {


can two run() execute concurrently? ie could it be that while we atomically switch pausedByUs to false here, the line 245 below gets executed? or is it all serialized by the awaitingTransaction and requested somehow ?

I don't believe so, the scheduler is a Schedulers.newSingle().

`ConsumerEventLoop.PollEvent` should not resume partitions that it, itself, has not paused. Prior to reactor@d908b81 pause/resume due to backpressure was protected by a boolean but it still unconditionally resumed all partitions. - reintroduce the boolean and only resume if we paused - keep track of user-paused partitions and exclude them from resuming

- enhance test to verify proper resumption after auto pause - continue to schedule polls while paused Now also resolves reactor#190

Fixes reactor#198 Only one `PollTask` needs to be scheduled at a time. Also upgrade docker image.

- `onRequest()` can be called on a different thread.

artembilan · 2021-03-10T15:25:01Z

src/main/java/reactor/kafka/receiver/internals/ConsumerEventLoop.java

@@ -116,6 +118,9 @@
    }

    void onRequest(long toAdd) {
+        if (log.isDebugEnabled()) {
+            log.debug("onRequest.toAdd:" + toAdd);


Probably need a space after colon to make it more readable in the logs

artembilan · 2021-03-10T15:27:19Z

src/main/java/reactor/kafka/receiver/internals/ConsumerEventLoop.java

                        consumer.pause(consumer.assignment());
+                        log.debug("Paused 2");


What the info does it bring for end-users in the logs?
I mean this one and the previous Paused 1 .
Doesn't look like very useful...

artembilan · 2021-03-10T15:33:39Z

src/test/java/reactor/kafka/AbstractKafkaTest.java

 public abstract class AbstractKafkaTest {

    public static final int DEFAULT_TEST_TIMEOUT = 60_000;

-    private static final KafkaContainer KAFKA = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:5.5.1"))
+    private static final KafkaContainer KAFKA = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:6.1.0"))


Can't we use a latest image here?
I mean is it possible to have it without version at all, which means latest?

artembilan · 2021-03-10T16:56:41Z

reactor-kafka-tools/src/test/resources/log4j.properties

@@ -0,0 +1,29 @@
+# Copyright (c) 2016-2018 Pivotal Software Inc, All Rights Reserved.


It's kinda weird to have a Copyright in the non-Java files, but OK if it is what we have so far in this project.
Anyway: I see you have modified Copyright to VMware in other place. So, this one might be affected, too.

On the other hand the Log4J 1.x is out of life for a while already: https://logging.apache.org/log4j/1.2/ 😄

Different issue though...

Oops - I changed the wrong one - intended to just change the new one. Need to figure out if they all need changing.

Yes; I am going to open another issue about log4j.

ivyazmitinov · 2021-03-10T21:44:19Z

src/main/java/reactor/kafka/receiver/internals/ConsumerEventLoop.java

-                        if (r > 1 || commitEvent.inProgress.get() > 0) {
-                            schedule();
-                        }
+                        schedule();
                    }

                    Operators.produced(REQUESTED, ConsumerEventLoop.this, 1);


Since poll returns empty ConsumerRecords for paused partitions, sink's queue quickly gets overflowed with garbage references, what could lead to the OOM, since empty ConsumerRecords is effectively added on each schedule call. Also, such logic breaks backpressure logic as empty ConsumerRecords is treated as satisfying demand from the downstream when it is clearly isn't.

In order to fix this it is enough to replace

Operators.produced(REQUESTED, ConsumerEventLoop.this, 1); sink.emitNext(records, ConsumerEventLoop.this);

With

if (!records.isEmpty()) { Operators.produced(REQUESTED, ConsumerEventLoop.this, 1); log.debug("Emmiting {} records", records.count()); sink.emitNext(records, ConsumerEventLoop.this); }

I had that before, but restored it because onRequest() was never called (it's called downstream from the sink) so we never resumed the consumer.

That might have been on an earlier iteration of this patch; I'll look again tomorrow.

See #183 (comment)

After further review, I had to pull that change because KafkaReceiverTests.userPause() now fails, because now there is no thread to call onRequest() to resume the consumer. Need to give this some more thought.

I was mistaken; that code looks fine; there is something else going on that I need to track down tomorrow.

garyrussell added the type/bug A general bug label Mar 3, 2021

garyrussell requested a review from simonbasle March 3, 2021 22:37

artembilan approved these changes Mar 3, 2021

View reviewed changes

simonbasle reviewed Mar 4, 2021

View reviewed changes

This was referenced Mar 4, 2021

Need help on backpressure and auto pause / resume during high load for reactive consumer #190

Closed

KafkaReceiver memory leak #198

Closed

garyrussell force-pushed the pause branch from 2b31272 to d615f62 Compare March 5, 2021 16:00

garyrussell mentioned this pull request Mar 10, 2021

Out of memory on DefaultKafkaReceiver with high batch size, returned by poll #183

Closed

garyrussell added 4 commits March 10, 2021 10:15

Address PR Comments

25260c5

- enhance test to verify proper resumption after auto pause - continue to schedule polls while paused Now also resolves reactor#190

Fix memory leak.

d128af4

Fixes reactor#198 Only one `PollTask` needs to be scheduled at a time. Also upgrade docker image.

Use AtomicBoolean for scheduled flag.

60fd644

- `onRequest()` can be called on a different thread.

garyrussell force-pushed the pause branch from 57bf869 to 60fd644 Compare March 10, 2021 15:15

artembilan requested changes Mar 10, 2021

View reviewed changes

Address PR Comments.

bf1275e

artembilan requested changes Mar 10, 2021

View reviewed changes

Fix copyrights.

031bc4a

artembilan approved these changes Mar 10, 2021

View reviewed changes

ivyazmitinov suggested changes Mar 10, 2021

View reviewed changes

garyrussell force-pushed the pause branch from a3365f1 to 031bc4a Compare March 10, 2021 22:39

ivyazmitinov mentioned this pull request Mar 11, 2021

Memory usage patch #203

Merged

Don't emit empty ConsumerRecords.

46aafa5

garyrussell merged commit 9fddabb into reactor:master Mar 11, 2021

garyrussell deleted the pause branch March 12, 2021 15:09

garyrussell mentioned this pull request May 18, 2022

Manually stopped partitions should not be resumed by reactor-kafka #48

Closed

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix Partition Resumes #199

Fix Partition Resumes #199

garyrussell commented Mar 3, 2021 •

edited

Loading

garyrussell commented Mar 3, 2021

artembilan commented Mar 3, 2021

simonbasle left a comment

simonbasle Mar 4, 2021

garyrussell Mar 4, 2021

simonbasle Mar 4, 2021

garyrussell Mar 4, 2021

simonbasle Mar 4, 2021

garyrussell Mar 4, 2021

artembilan Mar 10, 2021

artembilan Mar 10, 2021

artembilan Mar 10, 2021

artembilan Mar 10, 2021

garyrussell Mar 10, 2021

ivyazmitinov Mar 10, 2021

garyrussell Mar 10, 2021

garyrussell Mar 10, 2021

garyrussell Mar 10, 2021

garyrussell Mar 10, 2021

peterfigure Mar 11, 2021

		consumer.pause(consumer.assignment());
		log.debug("Paused 2");

		@@ -0,0 +1,29 @@
		# Copyright (c) 2016-2018 Pivotal Software Inc, All Rights Reserved.

Fix Partition Resumes #199

Fix Partition Resumes #199

Conversation

garyrussell commented Mar 3, 2021 • edited Loading

garyrussell commented Mar 3, 2021

artembilan commented Mar 3, 2021

simonbasle 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

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

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

garyrussell commented Mar 3, 2021 •

edited

Loading