lucasbru commented on code in PR #14226:
URL: https://github.com/apache/kafka/pull/14226#discussion_r1326415866
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -590,6 +590,30 @@ public void prepareRecycle() {
log.info("Closed and recycled state");
}
+ @Override
+ public void preparePoll() {
Review Comment:
Name: good idea. I wanted to use this as a "bag of things to do
before/after" the poll, as we do with prepareCommit / postCommit, but you are
right that it's not necessarily a good pattern, and we only have one thing in
the bag, so let's use a specific name.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -590,6 +590,30 @@ public void prepareRecycle() {
log.info("Closed and recycled state");
}
+ @Override
+ public void preparePoll() {
+ if (state() == State.RUNNING) {
+ // before running a new poll on the main consumer, if a queue's
buffered size has been
+ // decreased to the threshold, we can then resume the consumption
on this partition
+ final Set<TopicPartition> currentAssignment =
mainConsumer.assignment();
+
+ mainConsumer.resume(
+ partitionGroup.partitions()
+ .stream()
+ .filter(currentAssignment::contains)
+ .filter(p -> partitionGroup.numBuffered(p) <=
maxBufferedSize)
+ .collect(Collectors.toList())
+ );
+ }
+ }
+
+ @Override
+ public void postPoll() {
Review Comment:
Good idea. Done.
--
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
