Handle empty topics for the ProgressMonitor#356
Handle empty topics for the ProgressMonitor#356HenryCaiHaiying merged 4 commits intopinterest:masterfrom strava:fix_missing_topic
Conversation
| response.errorCode(topicPartition.getTopic(), topicPartition.getPartition())); | ||
| int errorCode = response.errorCode(topicPartition.getTopic(), topicPartition.getPartition()); | ||
|
|
||
| if (errorCode == 1) { |
There was a problem hiding this comment.
See response codes from the Kafka docs:
There was a problem hiding this comment.
Instead of hard-code 1, can we use the ErrorCodes (OutOrRange) from kafka side?
There was a problem hiding this comment.
Ah yeah that's a good idea. Will do.
| // That is no an exceptional situation - in fact it can be normal if | ||
| // the topic being consumed by Secor has a low volume. So in that | ||
| // case, simply return null | ||
| return null; |
There was a problem hiding this comment.
I also think a suitable alternative to catching this error here is to just let it bubble up, and catch it in the ProgressMonitor instead. Thoughts?
There was a problem hiding this comment.
I am fine with either way. The caller either needs to deal with that special exception or check null return value.
| response.errorCode(topicPartition.getTopic(), topicPartition.getPartition())); | ||
| int errorCode = response.errorCode(topicPartition.getTopic(), topicPartition.getPartition()); | ||
|
|
||
| if (errorCode == 1) { |
There was a problem hiding this comment.
Instead of hard-code 1, can we use the ErrorCodes (OutOrRange) from kafka side?
| // | ||
| // That is no an exceptional situation - in fact it can be normal if | ||
| // the topic being consumed by Secor has a low volume. So in that | ||
| // case, simply return null |
There was a problem hiding this comment.
Needs to have Log.WARN in this case.
There was a problem hiding this comment.
Sure thing, will add.
| // That is no an exceptional situation - in fact it can be normal if | ||
| // the topic being consumed by Secor has a low volume. So in that | ||
| // case, simply return null | ||
| return null; |
There was a problem hiding this comment.
I am fine with either way. The caller either needs to deal with that special exception or check null return value.
|
|
||
| long offsetLag = lastOffset - committedOffset; | ||
| long timestampMillisLag = lastTimestampMillis - committedTimestampMillis; | ||
| long offsetLag = 0L; |
There was a problem hiding this comment.
Should the code just return around line 195?
If committed message is null, what will the value of lastMessage? Will that also be null? Otherwise secor would have another upload after the committed message.
There was a problem hiding this comment.
Hmm...perhaps not return, but it could continue to skip this TopicPartition?
|
Alright, changes have been made, including the |
|
Thanks for the contribution |
2 participants
If a
TopicPartitionhas had all of its messages on the broker compacted away, when you callKafakaClient.getCommittedMessagewith thatTopicPartitionaRuntimeExceptionwas raised. This is due to the obvious fact thatgetCommittedMessageis attempting to return aMessage, however there is noMessageto return from the broker.This is not, I would argue, an exceptional situation. If the topic has low volume or an aggressive compaction configuration, often times the log on the broker will be empty. This generally does not cause problems, except in the case of #189, where the
ProgressMonitorwill crash on the error.This is not ideal, as actual progress is totally caught up. So this change attempts to address this situation through two changes:
KafakaClient.getCommittedMessageto returnnullif there is no message found.ProgressMonitor.getStatsto record a0lag value if no committed message was found.I've left some self-review notes below please take a look.
I'll also add that I didn't see any existing tests for the
ProgressMonitororKafakaClientso I did not add any. Let me know if you think it would be worth adding some however.