Read ExplicitLAC in asyncReadLastConfirmed - enable ExplicitLAC in new API

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

@@ -1383,6 +1383,15 @@ synchronized void updateLastConfirmed(long lac, long len) {
      */
 
     public void asyncReadLastConfirmed(final ReadLastConfirmedCallback cb, final Object ctx) {
+        if (clientCtx.getConf().useV2WireProtocol) {
+            // in v2 protocol we don't support readLAC RPC
+            asyncReadPiggybackLastConfirmed(cb, ctx);
+        } else {
+            asyncReadExplicitLastConfirmed(cb, ctx);
+        }
+    }
+
+    private void asyncReadPiggybackLastConfirmed(final ReadLastConfirmedCallback cb, final Object ctx) {
         boolean isClosed;
         long lastEntryId;
         synchronized (this) {

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

@@ -238,6 +238,7 @@ public ByteBufAllocator getByteBufAllocator() {
         setupBookieWatcherForNewEnsemble();
         setupBookieWatcherForEnsembleChange();
         setupBookieClientReadEntry();
+        setupBookieClientReadLac();
         setupBookieClientAddEntry();
         setupBookieClientForceLedger();
     }
@@ -524,6 +525,33 @@ protected void setupBookieClientReadEntry() {
                 any(), anyInt(), any(), anyBoolean());
     }
 
+    @SuppressWarnings("unchecked")
+    protected void setupBookieClientReadLac() {
+        final Stubber stub = doAnswer(invokation -> {
+            Object[] args = invokation.getArguments();
+            BookieId bookieSocketAddress = (BookieId) args[0];
+            long ledgerId = (Long) args[1];
+            final BookkeeperInternalCallbacks.ReadLacCallback callback =
+                (BookkeeperInternalCallbacks.ReadLacCallback) args[2];
+            Object ctx = args[3];
+            long entryId = BookieProtocol.LAST_ADD_CONFIRMED;
+            // simply use "readEntry" with LAST_ADD_CONFIRMED to get current LAC
+            // there is nothing that writes ExplicitLAC within MockBookKeeperTestCase
+            bookieClient.readEntry(bookieSocketAddress, ledgerId, entryId,
+                    new BookkeeperInternalCallbacks.ReadEntryCallback() {
+                @Override
+                public void readEntryComplete(int rc, long ledgerId, long entryId, ByteBuf buffer, Object ctx) {
+                    callback.readLacComplete(rc, ledgerId, null, buffer, ctx);
+                }
+            }, ctx, BookieProtocol.FLAG_NONE);
+            return null;
+        });
+
+        stub.when(bookieClient).readLac(any(BookieId.class), anyLong(),
+                any(BookkeeperInternalCallbacks.ReadLacCallback.class),
+                any());
+    }
+
     private byte[] extractEntryPayload(long ledgerId, long entryId, ByteBufList toSend)
             throws BKException.BKDigestMatchException {
         ByteBuf toSendCopy = Unpooled.copiedBuffer(toSend.toArray());

bookkeeper-server/src/test/java/org/apache/bookkeeper/client/api/ExplicitLACWithWriteHandleAPITest.java

@@ -0,0 +1,83 @@
+/**
+ *
+ * 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.api;
+
+import java.util.concurrent.TimeUnit;
+import org.apache.bookkeeper.conf.ClientConfiguration;
+import org.apache.bookkeeper.test.BookKeeperClusterTestCase;
+import org.apache.bookkeeper.util.TestUtils;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Tests about ExplicitLAC and {@link Handle} API.
+ */
+public class ExplicitLACWithWriteHandleAPITest extends BookKeeperClusterTestCase {
+
+    private static final Logger LOG = LoggerFactory.getLogger(ExplicitLACWithWriteHandleAPITest.class);
+
+    public ExplicitLACWithWriteHandleAPITest() {
+        super(1);
+    }
+
+    @Test
+    public void testUseExplicitLAC() throws Exception {
+        ClientConfiguration conf = new ClientConfiguration(baseClientConf);
+        conf.setExplictLacInterval(1000);
+        try (BookKeeper bkc = BookKeeper
+                .newBuilder(conf)
+                .build();) {
+            try (WriteHandle writer = bkc.newCreateLedgerOp()
+                    .withAckQuorumSize(1)
+                    .withEnsembleSize(1)
+                    .withPassword(new byte[0])
+                    .withWriteQuorumSize(1)
+                    .execute()
+                    .get();) {
+                writer.append("foo".getBytes("utf-8"));
+                writer.append("foo".getBytes("utf-8"));
+                writer.append("foo".getBytes("utf-8"));
+                long expectedLastAddConfirmed = writer.append("foo".getBytes("utf-8"));
+
+                // since BK 4.12.0 the reader automatically uses ExplicitLAC
+                try (ReadHandle r = bkc.newOpenLedgerOp()
+                        .withRecovery(false)
+                        .withPassword(new byte[0])
+                        .withLedgerId(writer.getId())
+                        .execute()
+                        .get()) {
+                    TestUtils.assertEventuallyTrue("ExplicitLAC did not ork", () -> {
+                        try {
+                            long value = r.readLastAddConfirmed();
+                            LOG.info("current value " + value + " vs " + expectedLastAddConfirmed);
+                            return value == expectedLastAddConfirmed;
+                        } catch (Exception ex) {
+                            throw new RuntimeException(ex);
+                        }
+                    }, 30, TimeUnit.SECONDS);
+                }
+
+            }
+
+        }
+    }
+}