Repository: nifi Updated Branches: refs/heads/master 79a7014a9 -> 620d44686
NIFI-5194: Ensure that even if calling KafkaConsumer.resume() throws an Exception, that we still release the lock that we are holding Signed-off-by: Pierre Villard <[email protected]> This closes #2701. Project: http://git-wip-us.apache.org/repos/asf/nifi/repo Commit: http://git-wip-us.apache.org/repos/asf/nifi/commit/620d4468 Tree: http://git-wip-us.apache.org/repos/asf/nifi/tree/620d4468 Diff: http://git-wip-us.apache.org/repos/asf/nifi/diff/620d4468 Branch: refs/heads/master Commit: 620d44686041a24ce03eb5d864b128c0e61e9dc5 Parents: 79a7014 Author: Mark Payne <[email protected]> Authored: Mon May 14 15:50:50 2018 -0400 Committer: Pierre Villard <[email protected]> Committed: Tue May 15 11:52:55 2018 +0200 ---------------------------------------------------------------------- .../apache/nifi/processors/kafka/pubsub/ConsumerLease.java | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/nifi/blob/620d4468/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java ---------------------------------------------------------------------- diff --git a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java index 6946dc1..99ef239 100644 --- a/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java +++ b/nifi-nar-bundles/nifi-kafka-bundle/nifi-kafka-0-9-processors/src/main/java/org/apache/nifi/processors/kafka/pubsub/ConsumerLease.java @@ -178,10 +178,13 @@ public abstract class ConsumerLease implements Closeable, ConsumerRebalanceListe logger.debug("Resuming " + assignments); } } finally { - if (assignments != null) { - kafkaConsumer.resume(assignments); + try { + if (assignments != null) { + kafkaConsumer.resume(assignments); + } + } finally { + pollingLock.unlock(); } - pollingLock.unlock(); } }
