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.client;
  
  import static com.google.common.base.Preconditions.checkState;
  
  import com.google.common.annotations.VisibleForTesting;
  import java.security.GeneralSecurityException;
  import java.util.List;
  import java.util.Map;
  import java.util.NavigableMap;
  import java.util.Set;
  import java.util.TreeMap;
  import java.util.concurrent.CompletableFuture;
  import java.util.concurrent.RejectedExecutionException;
  import org.apache.bookkeeper.client.AsyncCallback.AddCallback;
  import org.apache.bookkeeper.client.AsyncCallback.CloseCallback;
  import org.apache.bookkeeper.client.AsyncCallback.ReadCallback;
  import org.apache.bookkeeper.client.AsyncCallback.ReadLastConfirmedCallback;
  import org.apache.bookkeeper.client.api.LedgerMetadata;
  import org.apache.bookkeeper.client.api.WriteFlag;
  import org.apache.bookkeeper.net.BookieId;
  import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.GenericCallback;
  import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.LedgerMetadataListener;
  import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.ReadEntryListener;
  import org.apache.bookkeeper.proto.BookkeeperInternalCallbacks.TimedGenericCallback;
  import org.apache.bookkeeper.versioning.Version;
  import org.apache.bookkeeper.versioning.Versioned;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * Read only ledger handle. This ledger handle allows you to
   * read from a ledger but not to write to it. It overrides all
   * the public write operations from LedgerHandle.
   * It should be returned for BookKeeper#openLedger operations.
   */
  class ReadOnlyLedgerHandle extends LedgerHandle implements LedgerMetadataListener {
      private static final Logger LOG = LoggerFactory.getLogger(ReadOnlyLedgerHandle.class);
  
      private Object metadataLock = new Object();
      private final NavigableMap<Long, List<BookieId>> newEnsemblesFromRecovery = new TreeMap<>();
  
      class MetadataUpdater implements Runnable {
  
          final Versioned<LedgerMetadata> newMetadata;
  
          MetadataUpdater(Versioned<LedgerMetadata> metadata) {
              this.newMetadata = metadata;
          }
  
          @Override
          public void run() {
              while (true) {
                  Versioned<LedgerMetadata> currentMetadata = getVersionedLedgerMetadata();
                  Version.Occurred occurred = currentMetadata.getVersion().compare(newMetadata.getVersion());
                  if (Version.Occurred.BEFORE == occurred) {
                      synchronized (ReadOnlyLedgerHandle.this) {
                          if (setLedgerMetadata(currentMetadata, newMetadata)) {
                              LOG.info("Updated ledger metadata for ledger {} to {}, version {}.",
                                       ledgerId, newMetadata.getValue().toSafeString(), newMetadata.getVersion());
                              break;
                          }
                      }
                  } else {
                      break;
                  }
              }
          }
  
          @Override
          public String toString() {
              return String.format("MetadataUpdater(%d)", ledgerId);
          }
      }
  
      ReadOnlyLedgerHandle(ClientContext clientCtx,
                           long ledgerId, Versioned<LedgerMetadata> metadata,
                           BookKeeper.DigestType digestType, byte[] password,
                           boolean watch)
              throws GeneralSecurityException, NumberFormatException {
          super(clientCtx, ledgerId, metadata, digestType, password, WriteFlag.NONE);
          if (watch) {
              clientCtx.getLedgerManager().registerLedgerMetadataListener(ledgerId, this);
          }
      }
  
      @Override
      public void close()
              throws InterruptedException, BKException {
          clientCtx.getLedgerManager().unregisterLedgerMetadataListener(ledgerId, this);
      }
  
      @Override
      public void asyncClose(CloseCallback cb, Object ctx) {
          clientCtx.getLedgerManager().unregisterLedgerMetadataListener(ledgerId, this);
          cb.closeComplete(BKException.Code.OK, this, ctx);
      }
  
      @Override
      public long addEntry(byte[] data) throws InterruptedException, BKException {
          return addEntry(data, 0, data.length);
      }
  
      @Override
      public long addEntry(byte[] data, int offset, int length)
              throws InterruptedException, BKException {
          LOG.error("Tried to add entry on a Read-Only ledger handle, ledgerid=" + ledgerId);
          throw BKException.create(BKException.Code.IllegalOpException);
      }
  
      @Override
      public void asyncAddEntry(final byte[] data, final AddCallback cb,
                                final Object ctx) {
          asyncAddEntry(data, 0, data.length, cb, ctx);
      }
  
      @Override
      public void asyncAddEntry(final byte[] data, final int offset, final int length,
                                final AddCallback cb, final Object ctx) {
          LOG.error("Tried to add entry on a Read-Only ledger handle, ledgerid=" + ledgerId);
          cb.addComplete(BKException.Code.IllegalOpException, this,
                         LedgerHandle.INVALID_ENTRY_ID, ctx);
      }
  
      @Override
      public void onChanged(long lid, Versioned<LedgerMetadata> newMetadata) {
          if (LOG.isDebugEnabled()) {
              LOG.debug("Received ledger metadata update on {} : {}", lid, newMetadata);
          }
          if (this.ledgerId != lid) {
              return;
          }
          if (null == newMetadata) {
              return;
          }
          Versioned<LedgerMetadata> currentMetadata = getVersionedLedgerMetadata();
          Version.Occurred occurred = currentMetadata.getVersion().compare(newMetadata.getVersion());
          if (LOG.isDebugEnabled()) {
              LOG.debug("Try to update metadata from {} to {} : {}",
                        currentMetadata, newMetadata, occurred);
          }
          if (Version.Occurred.BEFORE == occurred) { // the metadata is updated
              try {
                  clientCtx.getMainWorkerPool().executeOrdered(ledgerId, new MetadataUpdater(newMetadata));
              } catch (RejectedExecutionException ree) {
                  LOG.error("Failed on submitting updater to update ledger metadata on ledger {} : {}",
                          ledgerId, newMetadata);
              }
          }
      }
  
      @Override
      public String toString() {
          return String.format("ReadOnlyLedgerHandle(lid = %d, id = %d)", ledgerId, super.hashCode());
      }
  
      @Override
      protected void initializeWriteHandleState() {
          // Essentially a noop, we don't want to set up write handle state here for a ReadOnlyLedgerHandle
          explicitLacFlushPolicy = ExplicitLacFlushPolicy.VOID_EXPLICITLAC_FLUSH_POLICY;
      }
  
      @Override
      public void asyncReadLastEntry(ReadCallback cb, Object ctx) {
          asyncReadLastConfirmed(new ReadLastConfirmedCallback() {
              @Override
              public void readLastConfirmedComplete(int rc, long lastConfirmed, Object ctx) {
                  if (rc == BKException.Code.OK) {
                      if (lastConfirmed < 0) {
                          // Ledger was empty, so there is no last entry to read
                          cb.readComplete(BKException.Code.NoSuchEntryException, ReadOnlyLedgerHandle.this, null, ctx);
                      } else {
                          asyncReadEntriesInternal(lastConfirmed, lastConfirmed, cb, ctx, false);
                      }
                  } else {
                      LOG.error("ReadException in asyncReadLastEntry, ledgerId: {}, lac: {}, rc:{}",
                          lastConfirmed, ledgerId, rc);
                      cb.readComplete(rc, ReadOnlyLedgerHandle.this, null, ctx);
                  }
              }
          }, ctx);
      }
  
      /**
       * For a read only ledger handle, this method will only ever be called during recovery,
       * when we are reading forward from LAC and writing back those entries. As such,
       * unlike with LedgerHandle, we do not want to persist changes to the metadata as they occur,
       * but rather, we want to defer the persistence until recovery has completed, and do it all
       * on the close.
       */
      @Override
      void handleBookieFailure(final Map<Integer, BookieId> failedBookies) {
          // handleBookieFailure should always run in the ordered executor thread for this
          // ledger, so this synchronized should be unnecessary, but putting it here now
          // just in case (can be removed when we validate threads)
          synchronized (metadataLock) {
              String logContext = String.format("[RecoveryEnsembleChange(ledger:%d)]", ledgerId);
  
              long lac = getLastAddConfirmed();
              LedgerMetadata metadata = getLedgerMetadata();
              List<BookieId> currentEnsemble = getCurrentEnsemble();
              try {
                  List<BookieId> newEnsemble = EnsembleUtils.replaceBookiesInEnsemble(
                          clientCtx.getBookieWatcher(), metadata, currentEnsemble, failedBookies, logContext);
                  Set<Integer> replaced = EnsembleUtils.diffEnsemble(currentEnsemble, newEnsemble);
                  if (!replaced.isEmpty()) {
                      newEnsemblesFromRecovery.put(lac + 1, newEnsemble);
                      unsetSuccessAndSendWriteRequest(newEnsemble, replaced);
                  }
              } catch (BKException.BKNotEnoughBookiesException e) {
                  LOG.error("Could not get additional bookie to remake ensemble, closing ledger: {}", ledgerId);
  
                  handleUnrecoverableErrorDuringAdd(e.getCode());
                  return;
              }
          }
      }
  
      @Override
      void handleUnrecoverableErrorDuringAdd(int rc) {
          errorOutPendingAdds(rc);
      }
  
      void recover(GenericCallback<Void> finalCb) {
          recover(finalCb, null, false);
      }
  
      /**
       * Recover the ledger.
       *
       * @param finalCb
       *          callback after recovery is done.
       * @param listener
       *          read entry listener on recovery reads.
       * @param forceRecovery
       *          force the recovery procedure even the ledger metadata shows the ledger is closed.
       */
      void recover(GenericCallback<Void> finalCb,
                   final @VisibleForTesting ReadEntryListener listener,
                   final boolean forceRecovery) {
          final GenericCallback<Void> cb = new TimedGenericCallback<Void>(
              finalCb,
              BKException.Code.OK,
              clientCtx.getClientStats().getRecoverOpLogger());
  
          MetadataUpdateLoop.NeedsUpdatePredicate needsUpdate =
              (metadata) -> metadata.getState() == LedgerMetadata.State.OPEN;
          if (forceRecovery) {
              // in the force recovery case, we want to update the metadata
              // to IN_RECOVERY, even if the ledger is already closed
              needsUpdate = (metadata) -> metadata.getState() != LedgerMetadata.State.IN_RECOVERY;
          }
          new MetadataUpdateLoop(
                  clientCtx.getLedgerManager(), getId(),
                  this::getVersionedLedgerMetadata,
                  needsUpdate,
                  (metadata) -> LedgerMetadataBuilder.from(metadata).withInRecoveryState().build(),
                  this::setLedgerMetadata)
              .run()
              .thenCompose((metadata) -> {
                      if (metadata.getValue().isClosed()) {
                          return CompletableFuture.completedFuture(ReadOnlyLedgerHandle.this);
                      } else {
                          return new LedgerRecoveryOp(ReadOnlyLedgerHandle.this, clientCtx)
                              .setEntryListener(listener)
                              .initiate();
                      }
              })
              .thenCompose((ignore) -> closeRecovered())
              .whenComplete((ignore, ex) -> {
                      if (ex != null) {
                          cb.operationComplete(
                                  BKException.getExceptionCode(ex, BKException.Code.UnexpectedConditionException), null);
                      } else {
                          cb.operationComplete(BKException.Code.OK, null);
                      }
              });
      }
  
      CompletableFuture<Versioned<LedgerMetadata>> closeRecovered() {
          long lac, len;
          synchronized (this) {
              lac = lastAddConfirmed;
              len = length.get();
          }
          LOG.info("Closing recovered ledger {} at entry {}", getId(), lac);
          CompletableFuture<Versioned<LedgerMetadata>> f = new MetadataUpdateLoop(
                  clientCtx.getLedgerManager(), getId(),
                  this::getVersionedLedgerMetadata,
                  (metadata) -> metadata.getState() == LedgerMetadata.State.IN_RECOVERY,
                  (metadata) -> {
                      LedgerMetadataBuilder builder = LedgerMetadataBuilder.from(metadata);
                      Long lastEnsembleKey = LedgerMetadataUtils.getLastEnsembleKey(metadata);
                      synchronized (metadataLock) {
                          newEnsemblesFromRecovery.entrySet().forEach(
                                  (e) -> {
                                      checkState(e.getKey() >= lastEnsembleKey,
                                                 "Once a ledger is in recovery, noone can add ensembles without closing");
                                      // Occurs when a bookie need to be replaced at very start of recovery
                                      if (lastEnsembleKey.equals(e.getKey())) {
                                          builder.replaceEnsembleEntry(e.getKey(), e.getValue());
                                      } else {
                                          builder.newEnsembleEntry(e.getKey(), e.getValue());
                                      }
                                  });
                      }
                      return builder.withClosedState().withLastEntryId(lac).withLength(len).build();
                  },
                  this::setLedgerMetadata).run();
          f.whenComplete((result, exception) -> {
              synchronized (metadataLock) {
                  newEnsemblesFromRecovery.clear();
              }
              if (exception != null) {
                  LOG.error("When closeRecovered,failed on clearing newEnsemblesFromRecovery.", exception);
              }
          });
          return f;
      }
  
      @Override
      List<BookieId> getCurrentEnsemble() {
          synchronized (metadataLock) {
              if (!newEnsemblesFromRecovery.isEmpty()) {
                  return newEnsemblesFromRecovery.lastEntry().getValue();
              } else {
                  return super.getCurrentEnsemble();
              }
          }
      }
  
  }
  