this.checkpointManager = checkpointManagerFactory.create(KafkaTopic.getTopic(),
KafkaLogWriterPlugin.CHECKPOINT_ROW_KEY_PREFIX);
this.partitioner = partitioner;
}
@Override
<BUG>public void getLogNext(final LoggingContext loggingContext, final LogOffset fromOffset, final int maxEvents,
</BUG>
final Filter filter, final Callback callback) {
<BUG>if (fromOffset != LogOffset.LATEST_OFFSET) {
long checkpointTime = getCheckpointTime(loggingContext);
if (fromOffset.getTime() < checkpointTime) {
fileLogReader.getLogNext(loggingContext, fromOffset, maxEvents, filter, callback);
</BUG>
if (callback.getCount() != 0) {

this.checkpointManager = checkpointManagerFactory.create(KafkaTopic.getTopic(),
KafkaLogWriterPlugin.CHECKPOINT_ROW_KEY_PREFIX);
this.partitioner = partitioner;
}
@Override
public void getLogNext(final LoggingContext loggingContext, final ReadRange readRange, final int maxEvents,
final Filter filter, final Callback callback) {
if (readRange != ReadRange.LATEST) {
long checkpointTime = getCheckpointTime(loggingContext);
if (readRange.getFromMillis() < checkpointTime) {
fileLogReader.getLogNext(loggingContext, readRange, maxEvents, filter, callback);
if (callback.getCount() != 0) {
