vamossagar12 commented on a change in pull request #11424:
URL: https://github.com/apache/kafka/pull/11424#discussion_r774025067
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java
##########
@@ -967,6 +980,17 @@ private long pollPhase() {
if (!records.isEmpty()) {
pollRecordsSensor.record(numRecords, now);
taskManager.addRecordsToTasks(records);
+ // Check buffer size after adding records to tasks
+ final long bufferSize = taskManager.getInputBufferSizeInBytes();
+ // Pausing partitions as the buffer size now exceeds max buffer
size
+ if (bufferSize > maxBufferSizeBytes.get()) {
+ log.info("Buffered records size {} bytes exceeds {}. Pausing
the consumer", bufferSize, maxBufferSizeBytes.get());
+ // Only non-empty partitions are paused here. Reason is that,
if a task has multiple partitions with
+ // some of them empty, then in that case pausing even empty
partitions would sacrifice ordered processing
+ // and even lead to temporal deadlock. More explanation can be
found here:
+ //
https://issues.apache.org/jira/browse/KAFKA-13152?focusedCommentId=17400647&page=com.atlassian.jira.plugin.system.issuetabpanels%3Acomment-tabpanel#comment-17400647
Review comment:
Oh that's because the ticket has a lot of comments and I intended to
point to the comment which talks about the design decision made here. If it
doesn't make sense, will remove it :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
