Unlock the write lock of the UnAckedMessageTracker before call redeli…

…verUnacknowledgedMessages (apache#10768)

The deadlock will happen in following steps:

1. The client internal thread got the consumer instance lock when call internalBatchReceiveAsync
2. The timer thread got the write lock of the UnAckedMessageTracker and then waiting on the consumer instance lock when call redeliverUnacknowledgedMessages
3. The client internal thread try to get the write lock of the UnAckedMessageTracker when adding unacked messages

The deadlock happens.

The fix is ensure the timer thread unlock the write lock before call redeliverUnacknowledgedMessages

pulsar-client/src/main/java/org/apache/pulsar/client/impl/UnAckedMessageTracker.java

@@ -139,15 +139,12 @@ public void run(Timeout t) throws Exception {
                     headPartition.clear();
                     timePartitions.addLast(headPartition);
                 } finally {
-                    try {
-                        if (messageIds.size() > 0) {
-                            consumerBase.onAckTimeoutSend(messageIds);
-                            consumerBase.redeliverUnacknowledgedMessages(messageIds);
-                        }
-                        timeout = client.timer().newTimeout(this, tickDurationInMs, TimeUnit.MILLISECONDS);
-                    } finally {
-                        writeLock.unlock();
+                    writeLock.unlock();
+                    if (messageIds.size() > 0) {
+                        consumerBase.onAckTimeoutSend(messageIds);
+                        consumerBase.redeliverUnacknowledgedMessages(messageIds);
                     }
+                    timeout = client.timer().newTimeout(this, tickDurationInMs, TimeUnit.MILLISECONDS);
                 }
             }
         }, this.tickDurationInMs, TimeUnit.MILLISECONDS);