MINOR: Leaves lock() outside the try block
Reviewers: Chia-Ping Tsai
@@ -254,8 +254,8 @@ public class RoundTripWorker implements TaskWorker {
spec.valueGenerator().generate(messageIndex));
producer.send(record, (metadata, exception) -> {
if (exception == null) {
- try {
lock.lock();
+ try {
unackedSends -= 1;
if (unackedSends <= 0)
unackedSendsAreZero.signalAll();
@@ -272,8 +272,8 @@ public class RoundTripWorker implements TaskWorker {
} catch (Throwable e) {
WorkerUtils.abort(log, "ProducerRunnable", e, doneFuture);
} finally {
- try {
lock.lock();
+ try {
log.info("{}: ProducerRunnable is exiting. messagesSent={}; uniqueMessagesSent={}; " +
"ackedSends={}/{}.", id, messagesSent, uniqueMessagesSent,
spec.maxMessages() - unackedSends, spec.maxMessages());
@@ -359,8 +359,8 @@ public class RoundTripWorker implements TaskWorker {
if (toReceiveTracker.removePending(messageIndex)) {
uniqueMessagesReceived++;
if (uniqueMessagesReceived >= spec.maxMessages()) {
- try {
lock.lock();
+ try {
log.info("{}: Consumer received the full count of {} unique messages. " +
"Waiting for all {} sends to be acked...", id, spec.maxMessages(), unackedSends);
while (unackedSends > 0)
