muralibasani commented on code in PR #22458:
URL: https://github.com/apache/kafka/pull/22458#discussion_r4383078030
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1244,6 +1279,9 @@ void runOnceWithoutProcessingThreads() {
final long pollLatency;
taskManager.resumePollingForPartitionsWithAvailableSpace();
Review Comment:
Agree, StreamThread now passes partitionsPausedForBufferOverflow into
resumePollingForPartitionsWithAvailableSpace, and StreamTask drops those
partitions from partitionsToResume before resuming.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java:
##########
@@ -1540,6 +1581,22 @@ private long pollPhase() {
taskManager.updateNextOffsets(records.nextOffsets());
}
+ // soft cap — read isn't atomic across tasks, so small overshoots are
fine.
+ final long bufferSize = taskManager.getInputBufferSizeInBytes();
+ if (maxBufferSizeInBytes.get() != UNDEFINED_INPUT_BUFFER_MAX_BYTES &&
bufferSize > maxBufferSizeInBytes.get()) {
+ // pause only non-empty partitions; pausing empty ones risks
ordering deadlock (KAFKA-13152).
+ // copy first — nonEmptyPartitions may be immutable, and pausing
an unassigned partition
+ // throws, so intersect with assignment (mirrors resume side).
+ final Set<TopicPartition> nonEmptyPartitions = new
HashSet<>(taskManager.nonEmptyPartitions());
+ nonEmptyPartitions.retainAll(mainConsumer.assignment());
+ if (!nonEmptyPartitions.isEmpty()) {
+ log.info("Buffered records size {} bytes exceeds {}. Pausing
partitions {} from the consumer",
Review Comment:
Done, both are DEBUG now.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
