Fix NPE issue during recovery add

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/LedgerHandle.java

@@ -1769,7 +1769,7 @@ void handleUnrecoverableErrorDuringAdd(int rc) {
         if (getLedgerMetadata().getState() == LedgerMetadata.State.IN_RECOVERY) {
             // we should not close ledger if ledger is recovery mode
             // otherwise we may lose entry.
-            errorOutPendingAdds(rc);
+            executeOrdered(() -> errorOutPendingAdds(rc));
             return;
         }
         LOG.error("Closing ledger {} due to {}", ledgerId, BKException.codeLogger(rc));

bookkeeper-server/src/main/java/org/apache/bookkeeper/client/ReadOnlyLedgerHandle.java

@@ -237,7 +237,7 @@ void handleBookieFailure(final Map<Integer, BookieId> failedBookies) {
 
     @Override
     void handleUnrecoverableErrorDuringAdd(int rc) {
-        errorOutPendingAdds(rc);
+        executeOrdered(() -> errorOutPendingAdds(rc));
     }
 
     void recover(GenericCallback<Void> finalCb) {

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockBookieWatcher.java

@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.bookkeeper.client;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.discover.RegistrationClient;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.proto.BookieAddressResolver;
+import org.apache.bookkeeper.stats.StatsLogger;
+
+public class MockBookieWatcher extends BookieWatcherImpl {
+
+    public MockBookieWatcher(ClientConfiguration conf, EnsemblePlacementPolicy placementPolicy,
+                             RegistrationClient registrationClient, BookieAddressResolver bookieAddressResolver,
+                             StatsLogger statsLogger) {
+        super(conf, placementPolicy, registrationClient, bookieAddressResolver, statsLogger);
+    }
+
+    @Override
+    public BookieId replaceBookie(int ensembleSize, int writeQuorumSize, int ackQuorumSize,
+                                  Map<String, byte[]> customMetadata, List<BookieId> existingBookies, int bookieIdx,
+                                  Set<BookieId> excludeBookies) throws BKException.BKNotEnoughBookiesException {
+        throw new BKException.BKNotEnoughBookiesException();
+    }
+}

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/MockClientContext.java

@@ -54,21 +54,14 @@ public class MockClientContext implements ClientContext {
     private MockRegistrationClient regClient;
     private ByteBufAllocator allocator;
 
-    static MockClientContext create(MockBookies mockBookies) throws Exception {
-        ClientConfiguration conf = new ClientConfiguration();
+    static MockClientContext create(MockBookies mockBookies, ClientConfiguration conf,
+                                    MockRegistrationClient regClient, EnsemblePlacementPolicy placementPolicy,
+                                    BookieWatcher bookieWatcher) throws Exception {
         OrderedScheduler scheduler = OrderedScheduler.newSchedulerBuilder().name("mock-executor").numThreads(1).build();
-        MockRegistrationClient regClient = new MockRegistrationClient();
-        EnsemblePlacementPolicy placementPolicy = new DefaultEnsemblePlacementPolicy();
-        BookieWatcherImpl bookieWatcherImpl = new BookieWatcherImpl(conf, placementPolicy,
-                                                                    regClient,
-                                                                    new DefaultBookieAddressResolver(regClient),
-                                                                    NullStatsLogger.INSTANCE);
-        bookieWatcherImpl.initialBlockingBookieRead();
-
         return new MockClientContext()
                 .setConf(ClientInternalConf.fromConfig(conf))
                 .setLedgerManager(new MockLedgerManager())
-                .setBookieWatcher(bookieWatcherImpl)
+                .setBookieWatcher(bookieWatcher)
                 .setPlacementPolicy(placementPolicy)
                 .setRegistrationClient(regClient)
                 .setBookieClient(new MockBookieClient(scheduler, mockBookies))
@@ -79,6 +72,18 @@ static MockClientContext create(MockBookies mockBookies) throws Exception {
                 .setIsClientClosed(() -> false);
     }
 
+    static MockClientContext create(MockBookies mockBookies) throws Exception {
+        ClientConfiguration conf = new ClientConfiguration();
+        MockRegistrationClient regClient = new MockRegistrationClient();
+        EnsemblePlacementPolicy placementPolicy = new DefaultEnsemblePlacementPolicy();
+        BookieWatcherImpl bookieWatcherImpl = new BookieWatcherImpl(conf, placementPolicy,
+                regClient,
+                new DefaultBookieAddressResolver(regClient),
+                NullStatsLogger.INSTANCE);
+        bookieWatcherImpl.initialBlockingBookieRead();
+        return create(mockBookies, conf, regClient, placementPolicy, bookieWatcherImpl);
+    }
+
     static MockClientContext create() throws Exception {
         MockBookies mockBookies = new MockBookies();
         return create(mockBookies);

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/PendingAddOpTest.java

@@ -25,14 +25,26 @@
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
+import com.google.common.collect.Lists;
 import io.netty.buffer.ByteBuf;
 import io.netty.buffer.Unpooled;
+import java.util.EnumSet;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.Callable;
 import java.util.concurrent.atomic.AtomicInteger;
 import org.apache.bookkeeper.client.BKException.Code;
+import org.apache.bookkeeper.client.api.LedgerMetadata;
 import org.apache.bookkeeper.client.api.WriteFlag;
 import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.discover.MockRegistrationClient;
+import org.apache.bookkeeper.net.BookieId;
+import org.apache.bookkeeper.net.BookieSocketAddress;
 import org.apache.bookkeeper.proto.BookieClient;
+import org.apache.bookkeeper.proto.MockBookies;
 import org.apache.bookkeeper.stats.NullStatsLogger;
+import org.apache.bookkeeper.versioning.Versioned;
 import org.junit.Before;
 import org.junit.Test;
 
@@ -60,6 +72,10 @@ public void setup() {
         lh = mock(LedgerHandle.class);
         when(lh.getDistributionSchedule())
             .thenReturn(new RoundRobinDistributionSchedule(3, 3, 2));
+        when(lh.getWriteFlags())
+                .thenReturn(EnumSet.of(WriteFlag.DEFERRED_SYNC));
+        Map<Integer, BookieId> failedBookies = new HashMap<>();
+        failedBookies.put(1, BookieId.parse("0.0.0.0:3181"));
         byte[] data = "test-pending-add-op".getBytes(UTF_8);
         payload = Unpooled.wrappedBuffer(data);
         payload.writerIndex(data.length);
@@ -87,4 +103,69 @@ public void testExecuteAfterCancelled() {
         assertNull(op.lh);
     }
 
+    @Test
+    public void testLedgerHandleWithDeferredSyncDuringRecoveryAdd() throws Exception {
+        final BookieId b1 = new BookieSocketAddress("b1", 3181).toBookieId();
+        final BookieId b2 = new BookieSocketAddress("b2", 3181).toBookieId();
+        final BookieId b3 = new BookieSocketAddress("b3", 3181).toBookieId();
+        MockClientContext clientCtx = MockClientContext.create();
+        Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 0,
+                LedgerMetadataBuilder.create().withInRecoveryState().newEnsembleEntry(0L,
+                        Lists.newArrayList(b1, b2, b3)));
+        LedgerHandle lh = new LedgerHandle(clientCtx, 0, md, BookKeeper.DigestType.CRC32C,
+                ClientUtil.PASSWD, EnumSet.of(WriteFlag.DEFERRED_SYNC));
+        lh.notifyWriteFailed(0, b1);
+        AtomicInteger rcHolder = new AtomicInteger(-0xdead);
+        PendingAddOp op = PendingAddOp.create(
+                lh, mockClientContext, lh.getCurrentEnsemble(),
+                payload, EnumSet.of(WriteFlag.DEFERRED_SYNC),
+                (rc, handle, entryId, qwcLatency, ctx) -> {
+                    rcHolder.set(rc);
+                }, null).enableRecoveryAdd();
+        assertSame(lh, op.lh);
+        op.setEntryId(0);
+        lh.pendingAddOps.add(op);
+        lh.clientCtx.getMainWorkerPool().submitOrdered(lh.ledgerId, (Callable<Void>) () -> {
+            op.run();
+            return null;
+        }).get();
+    }
+
+    @Test
+    public void testReadOnlyLedgerHandleWithNotEnoughBookiesExceptionDuringRecoveryAdd() throws Exception {
+        final BookieId b1 = new BookieSocketAddress("b1", 3181).toBookieId();
+        final BookieId b2 = new BookieSocketAddress("b2", 3181).toBookieId();
+        final BookieId b3 = new BookieSocketAddress("b3", 3181).toBookieId();
+        MockBookies mockBookies = new MockBookies();
+        ClientConfiguration conf = new ClientConfiguration();
+        MockRegistrationClient regClient = new MockRegistrationClient();
+        EnsemblePlacementPolicy placementPolicy = new DefaultEnsemblePlacementPolicy();
+        BookieWatcher bookieWatcher = new MockBookieWatcher(conf, placementPolicy,
+                regClient,
+                new DefaultBookieAddressResolver(regClient),
+                NullStatsLogger.INSTANCE);
+        ClientContext clientCtx = MockClientContext.create(mockBookies, conf, regClient, placementPolicy,
+                bookieWatcher);
+        Versioned<LedgerMetadata> md = ClientUtil.setupLedger(clientCtx, 0,
+                LedgerMetadataBuilder.create().withInRecoveryState().newEnsembleEntry(0L,
+                        Lists.newArrayList(b1, b2, b3)));
+        ReadOnlyLedgerHandle lh = new ReadOnlyLedgerHandle(clientCtx, 0, md, BookKeeper.DigestType.CRC32C,
+                ClientUtil.PASSWD, true);
+        lh.notifyWriteFailed(0, b1);
+        AtomicInteger rcHolder = new AtomicInteger(-0xdead);
+        PendingAddOp op = PendingAddOp.create(
+                lh, mockClientContext, lh.getCurrentEnsemble(),
+                payload, WriteFlag.NONE,
+                (rc, handle, entryId, qwcLatency, ctx) -> {
+                    rcHolder.set(rc);
+                }, null).enableRecoveryAdd();
+        assertSame(lh, op.lh);
+        op.setEntryId(0);
+        lh.pendingAddOps.add(op);
+        lh.clientCtx.getMainWorkerPool().submitOrdered(lh.ledgerId, (Callable<Void>) () -> {
+            op.run();
+            return null;
+        }).get();
+    }
+
 }