logstash shutdown does not release leases on blobs

[pawel-szon]

Logstash information:

1. Logstash version: logstash 8.1.0
2. Logstash installation source: expanded from logstash-8.1.0-linux-x86_64.tar.gz
3. How is Logstash being run: Via command line
4. How was the Logstash Plugin installed: came as part of logstash archive, then was updated to 1.4.4 using bin/logstash-plugin update logstash-input-azure_event_hubs

JVM: openjdk version "11.0.13" (bundled with logstash)

OS version:
Linux grafana-vm 5.4.17-2102.206.1.el8uek.x86_64 #2 SMP Wed Oct 6 16:24:16 PDT 2021 x86_64 x86_64 x86_64 GNU/Linux

Description of the problem including expected versus actual behavior:
When restarting logstash process (kill -TERM or ctrl+c and running it again) leases created on blobs holding information about position for each partition are sometimes not released and OWNINGHOST metadata remains to hold old event processor ID. Because of existing lease locks, Logstash is unable to read some partitions and therefore is not processing new events.
Expected behaviour would be to release all locks for all partition offset blobs on logstash shutdown.

Steps to reproduce:

1. Configure Azure Event Hub to have at least 2 partitions.
2. Configure logstash azure_event_hub input to read events from the Event Hub and use storage container for storing offset data.
3. Run logstash and make sure the events are processed correctly.
4. Stop logstash and check if all leases on offset blobs have been released (in our case it released offset file for only one partition).
5. Start logstash and notice that events from partition for which the lease has not been released are not being processed.

Logs attached.
restart.log