filter));
long latestOffset = kafkaConsumer.fetchOffsetBefore(KafkaConsumer.LATEST_OFFSET);
long earliestOffset = kafkaConsumer.fetchOffsetBefore(KafkaConsumer.EARLIEST_OFFSET);
long stopOffset;
long startOffset;
<BUG>if (fromOffset.getKafkaOffset() < 0)  {
</BUG>
stopOffset = latestOffset;
} else {
<BUG>stopOffset = fromOffset.getKafkaOffset();
</BUG>
}

filter));
long latestOffset = kafkaConsumer.fetchOffsetBefore(KafkaConsumer.LATEST_OFFSET);
long earliestOffset = kafkaConsumer.fetchOffsetBefore(KafkaConsumer.EARLIEST_OFFSET);
long stopOffset;
long startOffset;
if (readRange.getKafkaOffset() < 0)  {
stopOffset = latestOffset;
} else {
stopOffset = readRange.getKafkaOffset();
}
