public synchronized Consumer.Message nextMessage() {
try {
- if (fPendingMsgs.isEmpty()) {
+ if (!fPendingMsgs.isEmpty()) {
return makeMessage(fPendingMsgs.take());
}
} catch (InterruptedException x) {
log.warn("After size>0, pending msg take() threw InterruptedException. Ignoring. (" + x.getMessage() + ")",
x);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
Callable<Boolean> run = new Callable<Boolean>() {
curator.blockUntilConnected();
} catch (InterruptedException e) {
log.error("error while setting curator framework :",e);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
}
Thread.sleep(consumerHandoverWaitMs);
} catch (InterruptedException e) {
log.error("InterruptedException in dropTimedOutConsumer",e);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
log.info("Dropped " + key + " consumer due to timeout");
}
}
} catch (InterruptedException e) {
log.error("Exception "+ e);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
}
} catch (InterruptedException e) {
getLog().warn("Possible message loss. " + e.getMessage(), e);
- //Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
} catch (IOException e) {
getLog().warn("Possible message loss. " + e.getMessage(), e);
}
// FIXME: would be good to wait/signal
Thread.sleep(100);
} catch (InterruptedException e) {
- // Thread.currentThread().interrupt();
+ Thread.currentThread().interrupt();
}
}
});