ManagedLedger only closes ledger on error if current ledger (#240) (#…

…2573)

If we have a managed ledger, ml and we write 2 entries to it, if both
entries fail, both will end up calling ManagedLedgerImpl#ledgerClosed
with the ledger the write failed on as a parameter.

However, depending on timing, the second call to ledgerClosed could
end up adding a new ledger to the ledger list, even though the current
ledger is _not_ failing (as the failing ledger was replaced by the
first call).

This was the cause of a flake in
ManagedLedgerErrorsTest#recoverLongTimeAfterMultipleWriteErrors as
reported in (#240). However, it's not possible to get a deterministic
test for this as the timings need to be very precise. The failing
addComplete needs to run before first error handling completes, but
the runnable with ledgerClosed for the second failure needs to run
after the first error handling completes, but before the write resends
from the first error handling complete.

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerImpl.java

@@ -1277,7 +1277,8 @@ public synchronized void updateLedgersIdsComplete(Stat stat) {
 
     synchronized void ledgerClosed(final LedgerHandle lh) {
         final State state = STATE_UPDATER.get(this);
-        if (state == State.ClosingLedger || state == State.LedgerOpened) {
+        LedgerHandle currentLedger = this.currentLedger;
+        if (currentLedger == lh && (state == State.ClosingLedger || state == State.LedgerOpened)) {
             STATE_UPDATER.set(this, State.ClosedLedger);
         } else if (state == State.Closed) {
             // The managed ledger was closed during the write operation