Ensure that even if calling KafkaConsumer.resume() throws an Exception, that we still release the lock that we are holding
This closes
@@ -178,12 +178,15 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe
logger.debug("Resuming " + assignments);
}
} finally {
+ try {
if (assignments != null) {
kafkaConsumer.resume(assignments);
}
+ } finally {
pollingLock.unlock();
}
}
+ }
/**
* Notifies Kafka to commit the offsets for the specified topic/partition
