content: /*
   *
   * 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.bookie;
  
  import io.netty.buffer.ByteBuf;
  import java.io.IOException;
  import java.util.Arrays;
  import java.util.PrimitiveIterator.OfLong;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.atomic.AtomicBoolean;
  import org.apache.bookkeeper.client.api.BKException;
  import org.apache.bookkeeper.common.concurrent.FutureUtils;
  import org.apache.bookkeeper.common.util.Watcher;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Implements a ledger inside a bookie. In particular, it implements operations
   * to write entries to a ledger and read entries from a ledger.
   */
  public class LedgerDescriptorImpl extends LedgerDescriptor {
      private static final Logger LOG = LoggerFactory.getLogger(LedgerDescriptorImpl.class);
      final LedgerStorage ledgerStorage;
      private final long ledgerId;
      final byte[] masterKey;
  
      private final AtomicBoolean fenceEntryPersisted = new AtomicBoolean();
      private CompletableFuture<Boolean> logFenceResult = null;
  
      LedgerDescriptorImpl(byte[] masterKey,
                           long ledgerId,
                           LedgerStorage ledgerStorage) {
          this.masterKey = masterKey;
          this.ledgerId = ledgerId;
          this.ledgerStorage = ledgerStorage;
      }
  
      @Override
      void checkAccess(byte[] masterKey) throws BookieException, IOException {
          if (!Arrays.equals(this.masterKey, masterKey)) {
              LOG.error("[{}] Requested master key {} does not match the cached master key {}",
                      this.ledgerId, Arrays.toString(masterKey), Arrays.toString(this.masterKey));
              throw BookieException.create(BookieException.Code.UnauthorizedAccessException);
          }
      }
  
      @Override
      public long getLedgerId() {
          return ledgerId;
      }
  
      @Override
      boolean setFenced() throws IOException {
          return ledgerStorage.setFenced(ledgerId);
      }
  
      @Override
      boolean isFenced() throws IOException, BookieException {
          return ledgerStorage.isFenced(ledgerId);
      }
  
      @Override
      void setExplicitLac(ByteBuf lac) throws IOException {
          ledgerStorage.setExplicitLac(ledgerId, lac);
      }
  
      @Override
      ByteBuf getExplicitLac() throws IOException, BookieException {
          return ledgerStorage.getExplicitLac(ledgerId);
      }
  
      @Override
      synchronized CompletableFuture<Boolean> fenceAndLogInJournal(Journal journal) throws IOException {
          boolean success = this.setFenced();
          if (success) {
              // fenced for first time, we should add the key to journal ensure we can rebuild.
              return logFenceEntryInJournal(journal);
          } else {
              // If we reach here, it means the fence state in FileInfo has been set (may not be persisted yet).
              // However, writing the fence log entry to the journal might still be in progress. This can happen
              // when a bookie receives two fence requests almost at the same time. The subsequent logic is used
              // to check the fencing progress.
              if (logFenceResult == null || fenceEntryPersisted.get()){
                  // Either ledger's fenced state is recovered from Journal
                  // Or Log fence entry in Journal succeed
                  CompletableFuture<Boolean> result = FutureUtils.createFuture();
                  result.complete(true);
                  return result;
              } else if (logFenceResult.isDone()) {
                  // We failed to log fence entry in Journal, try again.
                  return logFenceEntryInJournal(journal);
              }
              // Fencing is in progress
              return logFenceResult;
          }
      }
  
      /**
       * Log the fence ledger entry in Journal so that we can rebuild the state.
       * @param journal log the fence entry in the Journal
       * @return A future which will be satisfied when add entry to journal complete
       */
      private CompletableFuture<Boolean> logFenceEntryInJournal(Journal journal) {
          CompletableFuture<Boolean> result;
          synchronized (this) {
              result = logFenceResult = FutureUtils.createFuture();
          }
          ByteBuf entry = createLedgerFenceEntry(ledgerId);
          try {
              journal.logAddEntry(entry, false /* ackBeforeSync */, (rc, ledgerId, entryId, addr, ctx) -> {
                  if (LOG.isDebugEnabled()) {
                      LOG.debug("Record fenced state for ledger {} in journal with rc {}",
                              ledgerId, BKException.codeLogger(rc));
                  }
                  if (rc == 0) {
                      fenceEntryPersisted.compareAndSet(false, true);
                      result.complete(true);
                  } else {
                      result.complete(false);
                  }
              }, null);
          } catch (InterruptedException e) {
              Thread.currentThread().interrupt();
              result.completeExceptionally(e);
          }
          return result;
      }
  
      @Override
      long addEntry(ByteBuf entry) throws IOException, BookieException {
          long ledgerId = entry.getLong(entry.readerIndex());
  
          if (ledgerId != this.ledgerId) {
              throw new IOException("Entry for ledger " + ledgerId + " was sent to " + this.ledgerId);
          }
  
          return ledgerStorage.addEntry(entry);
      }
  
      @Override
      ByteBuf readEntry(long entryId) throws IOException, BookieException {
          return ledgerStorage.getEntry(ledgerId, entryId);
      }
  
      @Override
      long getLastAddConfirmed() throws IOException, BookieException {
          return ledgerStorage.getLastAddConfirmed(ledgerId);
      }
  
      @Override
      boolean waitForLastAddConfirmedUpdate(long previousLAC,
                                            Watcher<LastAddConfirmedUpdateNotification> watcher) throws IOException {
          return ledgerStorage.waitForLastAddConfirmedUpdate(ledgerId, previousLAC, watcher);
      }
  
      @Override
      void cancelWaitForLastAddConfirmedUpdate(Watcher<LastAddConfirmedUpdateNotification> watcher) throws IOException {
          ledgerStorage.cancelWaitForLastAddConfirmedUpdate(ledgerId, watcher);
      }
  
      @Override
      OfLong getListOfEntriesOfLedger(long ledgerId) throws IOException {
          return ledgerStorage.getListOfEntriesOfLedger(ledgerId);
      }
  }
  