aliehsaeedii commented on code in PR #20693:
URL: https://github.com/apache/kafka/pull/20693#discussion_r2432365087
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamTask.java:
##########
@@ -788,9 +795,16 @@ record = partitionGroup.nextRecord(recordInfo,
wallClockTime);
consumedOffsets.put(partition, record.offset());
commitNeeded = true;
+ if (log.isTraceEnabled()) {
+ log.trace("Task {} processed record: topic={}, partition={},
offset={}",
+ id, record.topic(), record.partition(), record.offset());
+ }
+
// after processing this record, if its partition queue's buffered
size has been
// decreased to the threshold, we can then resume the consumption
on this partition
if (recordInfo.queue().size() <= maxBufferedSize) {
+ log.trace("Resume consumption for partition {}: buffered size
{} is under the threshold {}",
+ partition, recordInfo.queue().size(), maxBufferedSize);
partitionsToResume.add(partition);
}
Review Comment:
Does that make sense to add logs for catch-clauses as well?
--
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]
