Add logging to KafkaBasedLog::readToLogEnd
Simple logging additions at TRACE level that should help when the worker can't get caught up to the end of an internal topic.
Reviewers: Gwen Shapira Aakash Shah Konstantine Karantasis
@@ -281,9 +281,15 @@ public class KafkaBasedLog<K, V> {
Iterator<Map.Entry<TopicPartition, Long>> it = endOffsets.entrySet().iterator();
while (it.hasNext()) {
Map.Entry<TopicPartition, Long> entry = it.next();
- if (consumer.position(entry.getKey()) >= entry.getValue())
+ TopicPartition topicPartition = entry.getKey();
+ long endOffset = entry.getValue();
+ long lastConsumedOffset = consumer.position(topicPartition);
+ if (lastConsumedOffset >= endOffset) {
+ log.trace("Read to end offset {} for {}", endOffset, topicPartition);
it.remove();
- else {
+ } else {
+ log.trace("Behind end offset {} for {}; last-read offset is {}",
+ endOffset, topicPartition, lastConsumedOffset);
poll(Integer.MAX_VALUE);
break;
}
