Delay reading from the Kafka buffer as long as the circuit breaker is open #4135
Conversation
dlvenable
requested review from
chenqi0805,
engechas,
graytaylor0,
dinujoh,
kkondaka,
asifsmohammed and
oeyh
as code owners
| } else if (pauseConsumePredicate.pauseConsuming()) { | ||
| LOG.debug("Pause and skip the next consume from Kafka topic due to an external condition: {}", pauseConsumePredicate); | ||
| Thread.sleep(10000); | ||
| continue; |
There was a problem hiding this comment.
kafka consumer max.poll.interval will expire if we continue to pause for longer duration. This problem looks similar to #4023 where bufferAccumator gets full and we explicitly pause kafka consumer so that max poll does not expire. Not sure if we are taking in the PR (kkondaka). May be we can combine these 2 conditions and have a common solution.
There was a problem hiding this comment.
@hshardeesi , This is a good point. I'll take another look at that angle. Thanks!
There was a problem hiding this comment.
@hshardeesi , I pushed a change that pauses the Kafka consumer. I ended up calling it in a different location because this pause happens before any poll calls whereas the other happens while writing to the buffer.
There was a problem hiding this comment.
Good catch @hshardeesi I misunderstood it as part of shutdown handling.
There was a problem hiding this comment.
functionally this looks correct except that in other case (#4023) we don't pause consumer immediately but delay it until we are close to expiring kafka max.poll timeout. I feel this pause/resume logic should be abstracted out into separate class, similar to pausePredicate. abstraction can maintain state on when & what partitions to pause/resume etc and the core logic remains at one place.
There was a problem hiding this comment.
@hshardeesi , These are all good points. I agree with consolidating the logic to make it consistent. Do we need that in this PR, or can we follow this on?
… predicate is in place. This will allow the Kafka buffer to wait for the circuit breaker to close before reading. Signed-off-by: David Venable <[email protected]>
Signed-off-by: David Venable <[email protected]>
dlvenable
force-pushed
the
kafka-buffer-circuit-breaker
branch
from
b793536 to
12a636b
Compare
Signed-off-by: David Venable <[email protected]>
| consumer.pause(consumer.assignment()); | ||
| Thread.sleep(10000); | ||
| continue; | ||
| } else if(paused) { |
There was a problem hiding this comment.
Is this "else" required?
There was a problem hiding this comment.
I'm not sure it is strictly required. But, it avoids the extra call to resume() if we are not currently paused.
There was a problem hiding this comment.
@dlvenable that should not happen, right? We the resume is done, paused should be set to false. Looks like that's missing in your code.
| continue; | ||
| } else if(paused) { | ||
| LOG.debug("Resume consuming from Kafka topic."); | ||
| consumer.resume(consumer.assignment()); |
There was a problem hiding this comment.
I guess you missed setting paused = false here
There was a problem hiding this comment.
Good catch. I just pushed a fix.
Signed-off-by: David Venable <[email protected]>
Description
The
kafkabuffer was pausing 10 seconds to wait for the circuit breaker to close. However, it would then consume after that. This change will keep holding until the circuit breaker closes again. This is similar to how inputs into buffers work.Issues Resolved
N/A
Check List
By submitting this pull request, I confirm that my contribution is made under the terms of the Apache 2.0 license.
For more information on following Developer Certificate of Origin and signing off your commits, please check here.