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.activemq.store.kahadb;
  
  import java.io.DataInputStream;
  import java.io.IOException;
  import java.io.InterruptedIOException;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Iterator;
  import java.util.LinkedHashSet;
  import java.util.LinkedList;
  import java.util.List;
  import java.util.Map;
  import java.util.Map.Entry;
  import java.util.Optional;
  import java.util.Set;
  import java.util.concurrent.BlockingQueue;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.FutureTask;
  import java.util.concurrent.LinkedBlockingQueue;
  import java.util.concurrent.Semaphore;
  import java.util.concurrent.ThreadFactory;
  import java.util.concurrent.ThreadPoolExecutor;
  import java.util.concurrent.TimeUnit;
  import java.util.concurrent.TimeoutException;
  import java.util.concurrent.atomic.AtomicBoolean;
  import java.util.concurrent.atomic.AtomicInteger;
  import java.util.concurrent.atomic.AtomicLong;
  import java.util.concurrent.atomic.AtomicReference;
  
  import org.apache.activemq.broker.ConnectionContext;
  import org.apache.activemq.broker.region.BaseDestination;
  import org.apache.activemq.broker.scheduler.JobSchedulerStore;
  import org.apache.activemq.command.ActiveMQDestination;
  import org.apache.activemq.command.ActiveMQQueue;
  import org.apache.activemq.command.ActiveMQTempQueue;
  import org.apache.activemq.command.ActiveMQTempTopic;
  import org.apache.activemq.command.ActiveMQTopic;
  import org.apache.activemq.command.Message;
  import org.apache.activemq.command.MessageAck;
  import org.apache.activemq.command.MessageId;
  import org.apache.activemq.command.ProducerId;
  import org.apache.activemq.command.SubscriptionInfo;
  import org.apache.activemq.openwire.OpenWireFormat;
  import org.apache.activemq.protobuf.Buffer;
  import org.apache.activemq.store.AbstractMessageStore;
  import org.apache.activemq.store.IndexListener;
  import org.apache.activemq.store.ListenableFuture;
  import org.apache.activemq.store.MessageRecoveryContext;
  import org.apache.activemq.store.MessageRecoveryListener;
  import org.apache.activemq.store.MessageStore;
  import org.apache.activemq.store.MessageStoreStatistics;
  import org.apache.activemq.store.MessageStoreSubscriptionStatistics;
  import org.apache.activemq.store.NoLocalSubscriptionAware;
  import org.apache.activemq.store.PersistenceAdapter;
  import org.apache.activemq.store.ProxyMessageStore;
  import org.apache.activemq.store.TopicMessageStore;
  import org.apache.activemq.store.TransactionIdTransformer;
  import org.apache.activemq.store.TransactionStore;
  import org.apache.activemq.store.kahadb.data.KahaAddMessageCommand;
  import org.apache.activemq.store.kahadb.data.KahaDestination;
  import org.apache.activemq.store.kahadb.data.KahaDestination.DestinationType;
  import org.apache.activemq.store.kahadb.data.KahaLocation;
  import org.apache.activemq.store.kahadb.data.KahaRemoveDestinationCommand;
  import org.apache.activemq.store.kahadb.data.KahaRemoveMessageCommand;
  import org.apache.activemq.store.kahadb.data.KahaSubscriptionCommand;
  import org.apache.activemq.store.kahadb.data.KahaUpdateMessageCommand;
  import org.apache.activemq.store.kahadb.disk.journal.Location;
  import org.apache.activemq.store.kahadb.disk.page.Transaction;
  import org.apache.activemq.store.kahadb.disk.page.Transaction.CallableClosure;
  import org.apache.activemq.store.kahadb.disk.util.SequenceSet;
  import org.apache.activemq.store.kahadb.scheduler.JobSchedulerStoreImpl;
  import org.apache.activemq.usage.MemoryUsage;
  import org.apache.activemq.usage.SystemUsage;
  import org.apache.activemq.util.IOExceptionSupport;
  import org.apache.activemq.util.ServiceStopper;
  import org.apache.activemq.util.SubscriptionKey;
  import org.apache.activemq.util.ThreadPoolUtils;
  import org.apache.activemq.wireformat.WireFormat;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  public class KahaDBStore extends MessageDatabase implements PersistenceAdapter, NoLocalSubscriptionAware {
      static final Logger LOG = LoggerFactory.getLogger(KahaDBStore.class);
      private static final int MAX_ASYNC_JOBS = BaseDestination.MAX_AUDIT_DEPTH;
  
      public static final String PROPERTY_CANCELED_TASK_MOD_METRIC = "org.apache.activemq.store.kahadb.CANCELED_TASK_MOD_METRIC";
      public static final int cancelledTaskModMetric = Integer.parseInt(System.getProperty(
              PROPERTY_CANCELED_TASK_MOD_METRIC, "0"), 10);
      public static final String PROPERTY_ASYNC_EXECUTOR_MAX_THREADS = "org.apache.activemq.store.kahadb.ASYNC_EXECUTOR_MAX_THREADS";
      private static final int asyncExecutorMaxThreads = Integer.parseInt(System.getProperty(
              PROPERTY_ASYNC_EXECUTOR_MAX_THREADS, "1"), 10);
  
      protected ExecutorService queueExecutor;
      protected ExecutorService topicExecutor;
      protected final List<Map<AsyncJobKey, StoreTask>> asyncQueueMaps = new LinkedList<>();
      protected final List<Map<AsyncJobKey, StoreTask>> asyncTopicMaps = new LinkedList<>();
      final WireFormat wireFormat = new OpenWireFormat();
      private SystemUsage usageManager;
      private LinkedBlockingQueue<Runnable> asyncQueueJobQueue;
      private LinkedBlockingQueue<Runnable> asyncTopicJobQueue;
      Semaphore globalQueueSemaphore;
      Semaphore globalTopicSemaphore;
      private boolean concurrentStoreAndDispatchQueues = true;
      // when true, message order may be compromised when cache is exhausted if store is out
      // of order w.r.t cache
      private boolean concurrentStoreAndDispatchTopics = false;
      private int maxAsyncJobs = MAX_ASYNC_JOBS;
      private final KahaDBTransactionStore transactionStore;
      private TransactionIdTransformer transactionIdTransformer;
  
      public KahaDBStore() {
          this.transactionStore = new KahaDBTransactionStore(this);
          this.transactionIdTransformer = txid -> txid;
      }
  
      @Override
      public String toString() {
          return "KahaDB:[" + directory.getAbsolutePath() + "]";
      }
  
      @Override
      public void setBrokerName(String brokerName) {
      }
  
      @Override
      public void setUsageManager(SystemUsage usageManager) {
          this.usageManager = usageManager;
      }
  
      public SystemUsage getUsageManager() {
          return this.usageManager;
      }
  
      /**
       * @return the concurrentStoreAndDispatch
       */
      public boolean isConcurrentStoreAndDispatchQueues() {
          return this.concurrentStoreAndDispatchQueues;
      }
  
      /**
       * @param concurrentStoreAndDispatch
       *            the concurrentStoreAndDispatch to set
       */
      public void setConcurrentStoreAndDispatchQueues(boolean concurrentStoreAndDispatch) {
          this.concurrentStoreAndDispatchQueues = concurrentStoreAndDispatch;
      }
  
      /**
       * @return the concurrentStoreAndDispatch
       */
      public boolean isConcurrentStoreAndDispatchTopics() {
          return this.concurrentStoreAndDispatchTopics;
      }
  
      /**
       * @param concurrentStoreAndDispatch
       *            the concurrentStoreAndDispatch to set
       */
      public void setConcurrentStoreAndDispatchTopics(boolean concurrentStoreAndDispatch) {
          this.concurrentStoreAndDispatchTopics = concurrentStoreAndDispatch;
      }
  
      /**
       * @return the maxAsyncJobs
       */
      public int getMaxAsyncJobs() {
          return this.maxAsyncJobs;
      }
  
      /**
       * @param maxAsyncJobs
       *            the maxAsyncJobs to set
       */
      public void setMaxAsyncJobs(int maxAsyncJobs) {
          this.maxAsyncJobs = maxAsyncJobs;
      }
  
  
      @Override
      protected void configureMetadata() {
          if (brokerService != null) {
              metadata.openwireVersion = brokerService.getStoreOpenWireVersion();
              wireFormat.setVersion(metadata.openwireVersion);
  
              if (LOG.isDebugEnabled()) {
                  LOG.debug("Store OpenWire version configured as: {}", metadata.openwireVersion);
              }
  
          }
      }
  
      @Override
      public void doStart() throws Exception {
          //configure the metadata before start, right now
          //this is just the open wire version
          LOG.info("Starting KahaDBStore");
          configureMetadata();
          super.doStart();
  
          if (brokerService != null) {
              // In case the recovered store used a different OpenWire version log a warning
              // to assist in determining why journal reads fail.
              if (metadata.openwireVersion != brokerService.getStoreOpenWireVersion()) {
                  LOG.warn("Existing Store uses a different OpenWire version[{}] " +
                           "than the version configured[{}] reverting to the version " +
                           "used by this store, some newer broker features may not work" +
                           "as expected.",
                           metadata.openwireVersion, brokerService.getStoreOpenWireVersion());
  
                  // Update the broker service instance to the actual version in use.
                  wireFormat.setVersion(metadata.openwireVersion);
                  brokerService.setStoreOpenWireVersion(metadata.openwireVersion);
              }
          }
  
          this.globalQueueSemaphore = new Semaphore(getMaxAsyncJobs());
          this.globalTopicSemaphore = new Semaphore(getMaxAsyncJobs());
          this.asyncQueueJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs());
          this.asyncTopicJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs());
          this.queueExecutor = new StoreTaskExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
              asyncQueueJobQueue, new ThreadFactory() {
                  @Override
                  public Thread newThread(Runnable runnable) {
                      Thread thread = new Thread(runnable, "ConcurrentQueueStoreAndDispatch");
                      thread.setDaemon(true);
                      return thread;
                  }
              });
          this.topicExecutor = new StoreTaskExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
              asyncTopicJobQueue, new ThreadFactory() {
                  @Override
                  public Thread newThread(Runnable runnable) {
                      Thread thread = new Thread(runnable, "ConcurrentTopicStoreAndDispatch");
                      thread.setDaemon(true);
                      return thread;
                  }
              });
      }
  
      @Override
      public void doStop(ServiceStopper stopper) throws Exception {
          // drain down async jobs
          LOG.info("Stopping async queue tasks");
          if (this.globalQueueSemaphore != null) {
              this.globalQueueSemaphore.tryAcquire(this.maxAsyncJobs, 60, TimeUnit.SECONDS);
          }
          synchronized (this.asyncQueueMaps) {
              for (Map<AsyncJobKey, StoreTask> m : asyncQueueMaps) {
                  synchronized (m) {
                      for (StoreTask task : m.values()) {
                          task.cancel();
                      }
                  }
              }
              this.asyncQueueMaps.clear();
          }
          LOG.info("Stopping async topic tasks");
          if (this.globalTopicSemaphore != null) {
              this.globalTopicSemaphore.tryAcquire(this.maxAsyncJobs, 60, TimeUnit.SECONDS);
          }
          synchronized (this.asyncTopicMaps) {
              for (Map<AsyncJobKey, StoreTask> m : asyncTopicMaps) {
                  synchronized (m) {
                      for (StoreTask task : m.values()) {
                          task.cancel();
                      }
                  }
              }
              this.asyncTopicMaps.clear();
          }
          if (this.globalQueueSemaphore != null) {
              this.globalQueueSemaphore.drainPermits();
          }
          if (this.globalTopicSemaphore != null) {
              this.globalTopicSemaphore.drainPermits();
          }
          if (this.queueExecutor != null) {
              ThreadPoolUtils.shutdownNow(queueExecutor);
              queueExecutor = null;
          }
          if (this.topicExecutor != null) {
              ThreadPoolUtils.shutdownNow(topicExecutor);
              topicExecutor = null;
          }
          LOG.info("Stopped KahaDB");
          super.doStop(stopper);
      }
  
      private Location findMessageLocation(final String key, final KahaDestination destination) throws IOException {
          return pageFile.tx().execute(new Transaction.CallableClosure<Location, IOException>() {
              @Override
              public Location execute(Transaction tx) throws IOException {
                  StoredDestination sd = getStoredDestination(destination, tx);
                  Long sequence = sd.messageIdIndex.get(tx, key);
                  if (sequence == null) {
                      return null;
                  }
                  return sd.orderIndex.get(tx, sequence).location;
              }
          });
      }
  
      protected StoreQueueTask removeQueueTask(KahaDBMessageStore store, MessageId id) {
          StoreQueueTask task = null;
          synchronized (store.asyncTaskMap) {
              task = (StoreQueueTask) store.asyncTaskMap.remove(new AsyncJobKey(id, store.getDestination()));
          }
          return task;
      }
  
      // with asyncTaskMap locked
      protected void addQueueTask(KahaDBMessageStore store, StoreQueueTask task) throws IOException {
          store.asyncTaskMap.put(new AsyncJobKey(task.getMessage().getMessageId(), store.getDestination()), task);
          this.queueExecutor.execute(task);
      }
  
      protected StoreTopicTask removeTopicTask(KahaDBTopicMessageStore store, MessageId id) {
          StoreTopicTask task = null;
          synchronized (store.asyncTaskMap) {
              task = (StoreTopicTask) store.asyncTaskMap.remove(new AsyncJobKey(id, store.getDestination()));
          }
          return task;
      }
  
      protected void addTopicTask(KahaDBTopicMessageStore store, StoreTopicTask task) throws IOException {
          synchronized (store.asyncTaskMap) {
              store.asyncTaskMap.put(new AsyncJobKey(task.getMessage().getMessageId(), store.getDestination()), task);
          }
          this.topicExecutor.execute(task);
      }
  
      @Override
      public TransactionStore createTransactionStore() throws IOException {
          return this.transactionStore;
      }
  
      public boolean getForceRecoverIndex() {
          return this.forceRecoverIndex;
      }
  
      public void setForceRecoverIndex(boolean forceRecoverIndex) {
          this.forceRecoverIndex = forceRecoverIndex;
      }
  
      public void forgetRecoveredAcks(ArrayList<MessageAck> preparedAcks, boolean isRollback) throws IOException {
          if (preparedAcks != null) {
              Map<ActiveMQDestination, KahaDBMessageStore> stores = new HashMap<>();
              for (MessageAck ack : preparedAcks) {
                  stores.put(ack.getDestination(), findMatchingStore(ack.getDestination()));
              }
              ArrayList<MessageAck> perStoreAcks = new ArrayList<>();
              for (Entry<ActiveMQDestination, KahaDBMessageStore> entry : stores.entrySet()) {
                  for (MessageAck ack : preparedAcks) {
                      if (entry.getKey().equals(ack.getDestination())) {
                          perStoreAcks.add(ack);
                      }
                  }
                  entry.getValue().forgetRecoveredAcks(perStoreAcks, isRollback);
                  perStoreAcks.clear();
              }
          }
      }
  
      public void trackRecoveredAcks(ArrayList<MessageAck> preparedAcks) throws IOException {
          Map<ActiveMQDestination, KahaDBMessageStore> stores = new HashMap<>();
          for (MessageAck ack : preparedAcks) {
              stores.put(ack.getDestination(), findMatchingStore(ack.getDestination()));
          }
          ArrayList<MessageAck> perStoreAcks = new ArrayList<>();
          for (Entry<ActiveMQDestination, KahaDBMessageStore> entry : stores.entrySet()) {
              for (MessageAck ack : preparedAcks) {
                  if (entry.getKey().equals(ack.getDestination())) {
                      perStoreAcks.add(ack);
                  }
              }
              entry.getValue().trackRecoveredAcks(perStoreAcks);
              perStoreAcks.clear();
          }
      }
  
      private KahaDBMessageStore findMatchingStore(ActiveMQDestination activeMQDestination) throws IOException {
          ProxyMessageStore store = (ProxyMessageStore) storeCache.get(key(convert(activeMQDestination)));
          if (store == null) {
              if (activeMQDestination.isQueue()) {
                  store = (ProxyMessageStore) createQueueMessageStore((ActiveMQQueue) activeMQDestination);
              } else {
                  store = (ProxyMessageStore) createTopicMessageStore((ActiveMQTopic) activeMQDestination);
              }
          }
          return (KahaDBMessageStore) store.getDelegate();
      }
  
      public class KahaDBMessageStore extends AbstractMessageStore {
          protected final Map<AsyncJobKey, StoreTask> asyncTaskMap = new HashMap<AsyncJobKey, StoreTask>();
          protected KahaDestination dest;
          private final int maxAsyncJobs;
          private final Semaphore localDestinationSemaphore;
          protected final HashMap<String, Set<String>> ackedAndPreparedMap = new HashMap<String, Set<String>>();
          protected final HashMap<String, Set<String>> rolledBackAcksMap = new HashMap<String, Set<String>>();
  
          double doneTasks, canceledTasks = 0;
  
          public KahaDBMessageStore(ActiveMQDestination destination) {
              super(destination);
              this.dest = convert(destination);
              this.maxAsyncJobs = getMaxAsyncJobs();
              this.localDestinationSemaphore = new Semaphore(this.maxAsyncJobs);
          }
  
          @Override
          public ActiveMQDestination getDestination() {
              return destination;
          }
  
  
          private final String recoveredTxStateMapKey(ActiveMQDestination destination, MessageAck ack) {
              return destination.isQueue() ? destination.getPhysicalName() : ack.getConsumerId().getConnectionId();
          }
  
          // messages that have prepared (pending) acks cannot be re-dispatched unless the outcome is rollback,
          // till then they are skipped by the store.
          // 'at most once' XA guarantee
          public void trackRecoveredAcks(ArrayList<MessageAck> acks) {
              indexLock.writeLock().lock();
              try {
                  for (MessageAck ack : acks) {
                      final String key = recoveredTxStateMapKey(destination, ack);
                      Set<String> ackedAndPrepared = ackedAndPreparedMap.get(key);
                      if (ackedAndPrepared == null) {
                          ackedAndPrepared = new LinkedHashSet<String>();
                          ackedAndPreparedMap.put(key, ackedAndPrepared);
                      }
                      ackedAndPrepared.add(ack.getLastMessageId().toProducerKey());
                  }
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          public void forgetRecoveredAcks(ArrayList<MessageAck> acks, boolean rollback) throws IOException {
              if (acks != null) {
                  indexLock.writeLock().lock();
                  try {
                      for (MessageAck ack : acks) {
                          final String id = ack.getLastMessageId().toProducerKey();
                          final String key = recoveredTxStateMapKey(destination, ack);
                          Set<String> ackedAndPrepared = ackedAndPreparedMap.get(key);
                          if (ackedAndPrepared != null) {
                              ackedAndPrepared.remove(id);
                              if (ackedAndPreparedMap.isEmpty()) {
                                  ackedAndPreparedMap.remove(key);
                              }
                          }
                          if (rollback) {
                              Set<String> rolledBackAcks = rolledBackAcksMap.get(key);
                              if (rolledBackAcks == null) {
                                  rolledBackAcks = new LinkedHashSet<String>();
                                  rolledBackAcksMap.put(key, rolledBackAcks);
                              }
                              rolledBackAcks.add(id);
                              pageFile.tx().execute(tx -> {
                                  incrementAndAddSizeToStoreStat(tx, dest, 0);
                              });
                          }
                      }
                  } finally {
                      indexLock.writeLock().unlock();
                  }
              }
          }
  
          @Override
          public ListenableFuture<Object> asyncAddQueueMessage(final ConnectionContext context, final Message message)
                  throws IOException {
              if (isConcurrentStoreAndDispatchQueues()) {
                  message.beforeMarshall(wireFormat);
                  StoreQueueTask result = new StoreQueueTask(this, context, message);
                  ListenableFuture<Object> future = result.getFuture();
                  message.getMessageId().setFutureOrSequenceLong(future);
                  message.setRecievedByDFBridge(true); // flag message as concurrentStoreAndDispatch
                  result.aquireLocks();
                  synchronized (asyncTaskMap) {
                      addQueueTask(this, result);
                      if (indexListener != null) {
                          indexListener.onAdd(new IndexListener.MessageContext(context, message, null));
                      }
                  }
                  return future;
              } else {
                  return super.asyncAddQueueMessage(context, message);
              }
          }
  
          @Override
          public void removeAsyncMessage(ConnectionContext context, MessageAck ack) throws IOException {
              if (isConcurrentStoreAndDispatchQueues()) {
                  AsyncJobKey key = new AsyncJobKey(ack.getLastMessageId(), getDestination());
                  StoreQueueTask task = null;
                  synchronized (asyncTaskMap) {
                      task = (StoreQueueTask) asyncTaskMap.get(key);
                  }
                  if (task != null) {
                      if (ack.isInTransaction() || !task.cancel()) {
                          try {
                              task.future.get();
                          } catch (InterruptedException e) {
                              throw new InterruptedIOException(e.toString());
                          } catch (Exception ignored) {
                              LOG.debug("removeAsync: cannot cancel, waiting for add resulted in ex", ignored);
                          }
                          removeMessage(context, ack);
                      } else {
                          indexLock.writeLock().lock();
                          try {
                              metadata.producerSequenceIdTracker.isDuplicate(ack.getLastMessageId());
                          } finally {
                              indexLock.writeLock().unlock();
                          }
                          synchronized (asyncTaskMap) {
                              asyncTaskMap.remove(key);
                          }
                      }
                  } else {
                      removeMessage(context, ack);
                  }
              } else {
                  removeMessage(context, ack);
              }
          }
  
          @Override
          public void addMessage(final ConnectionContext context, final Message message) throws IOException {
              final KahaAddMessageCommand command = new KahaAddMessageCommand();
              command.setDestination(dest);
              command.setMessageId(message.getMessageId().toProducerKey());
              command.setTransactionInfo(TransactionIdConversion.convert(transactionIdTransformer.transform(message.getTransactionId())));
              command.setPriority(message.getPriority());
              command.setPrioritySupported(isPrioritizedMessages());
              org.apache.activemq.util.ByteSequence packet = wireFormat.marshal(message);
              command.setMessage(new Buffer(packet.getData(), packet.getOffset(), packet.getLength()));
              store(command, isEnableJournalDiskSyncs() && message.isResponseRequired(), new IndexAware() {
                  // sync add? (for async, future present from getFutureOrSequenceLong)
                  Object possibleFuture = message.getMessageId().getFutureOrSequenceLong();
  
                  @Override
                  public void sequenceAssignedWithIndexLocked(final long sequence) {
                      message.getMessageId().setFutureOrSequenceLong(sequence);
                      if (indexListener != null) {
                          if (possibleFuture == null) {
                              trackPendingAdd(dest, sequence);
                              indexListener.onAdd(new IndexListener.MessageContext(context, message, new Runnable() {
                                  @Override
                                  public void run() {
                                      trackPendingAddComplete(dest, sequence);
                                  }
                              }));
                          }
                      }
                  }
              }, null);
  
              /*
               * After we store the command in the journal we no longer need to keep the message
               * on the command, and we can clear the field here.
               *
               * The reason to clear the message is that for messages added as part of a transaction the command
               * will be added to the inflightTransactions map as a pending add operation.
               * For long-running transactions and/or transactions with a lot of pending messages
               * (or large messages) this can use up a decent amount of memory which can increase GC pressure.
               *
               * The commands are only tracked in the map so that the KahaDB index can be updated later
               * on transaction commit, but updating the index only requires metadata from the command
               * such as message id or destination and not the message itself, so we can safely clear the field.
               *
               * Note that on broker restart and recovery of the KahaDB journal the pending message
               * adds for transactions will be loaded again and the memory won't be cleared in that case.
               * This could be revisited in the future if an issue but that should not be a large
               * issue because that's only done on first startup and during recovery and then
               * after the broker is recovered the memory footprint will drop. Also, as of now, recovering
               * XA transactions in the transaction broker requires loading the messages and acks anyway
               * for processing, so we need to load the full message and keep it in the pending operation.
               */
              command.clearMessage();
          }
  
          @Override
          public void updateMessage(Message message) throws IOException {
              if (LOG.isTraceEnabled()) {
                  LOG.trace("updating: " + message.getMessageId() + " with deliveryCount: " + message.getRedeliveryCounter());
              }
              KahaUpdateMessageCommand updateMessageCommand = new KahaUpdateMessageCommand();
              KahaAddMessageCommand command = new KahaAddMessageCommand();
              command.setDestination(dest);
              command.setMessageId(message.getMessageId().toProducerKey());
              command.setPriority(message.getPriority());
              command.setPrioritySupported(prioritizedMessages);
              org.apache.activemq.util.ByteSequence packet = wireFormat.marshal(message);
              command.setMessage(new Buffer(packet.getData(), packet.getOffset(), packet.getLength()));
              updateMessageCommand.setMessage(command);
              store(updateMessageCommand, isEnableJournalDiskSyncs(), null, null);
          }
  
          @Override
          public void removeMessage(ConnectionContext context, MessageAck ack) throws IOException {
              KahaRemoveMessageCommand command = new KahaRemoveMessageCommand();
              command.setDestination(dest);
              command.setMessageId(ack.getLastMessageId().toProducerKey());
              command.setTransactionInfo(TransactionIdConversion.convert(transactionIdTransformer.transform(ack.getTransactionId())));
  
              org.apache.activemq.util.ByteSequence packet = wireFormat.marshal(ack);
              command.setAck(new Buffer(packet.getData(), packet.getOffset(), packet.getLength()));
              store(command, isEnableJournalDiskSyncs() && ack.isResponseRequired(), null, null);
          }
  
          @Override
          public void removeAllMessages(ConnectionContext context) throws IOException {
              KahaRemoveDestinationCommand command = new KahaRemoveDestinationCommand();
              command.setDestination(dest);
              store(command, true, null, null);
          }
  
          @Override
          public Message getMessage(MessageId identity) throws IOException {
              final String key = identity.toProducerKey();
  
              // Hopefully one day the page file supports concurrent read
              // operations... but for now we must
              // externally synchronize...
              Location location;
              indexLock.writeLock().lock();
              try {
                  location = findMessageLocation(key, dest);
              } finally {
                  indexLock.writeLock().unlock();
              }
              if (location == null) {
                  return null;
              }
  
              return loadMessage(location);
          }
  
          @Override
          public boolean isEmpty() throws IOException {
              indexLock.writeLock().lock();
              try {
                  return pageFile.tx().execute(new Transaction.CallableClosure<Boolean, IOException>() {
                      @Override
                      public Boolean execute(Transaction tx) throws IOException {
                          // Iterate through all index entries to get a count of
                          // messages in the destination.
                          StoredDestination sd = getStoredDestination(dest, tx);
                          return sd.locationIndex.isEmpty(tx);
                      }
                  });
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          @Override
          public void recover(final MessageRecoveryListener listener) throws Exception {
              // recovery may involve expiry which will modify
              indexLock.writeLock().lock();
              try {
                  pageFile.tx().execute(new Transaction.Closure<Exception>() {
                      @Override
                      public void execute(Transaction tx) throws Exception {
                          StoredDestination sd = getStoredDestination(dest, tx);
                          recoverRolledBackAcks(destination.getPhysicalName(), sd, tx, Integer.MAX_VALUE, listener);
                          sd.orderIndex.resetCursorPosition();
                          for (Iterator<Entry<Long, MessageKeys>> iterator = 
                                  sd.orderIndex.iterator(tx, new MessageOrderCursor()); listener.hasSpace() && 
                                  iterator.hasNext(); ) {
                              Entry<Long, MessageKeys> entry = iterator.next();
                              Set<String> ackedAndPrepared = ackedAndPreparedMap.get(destination.getPhysicalName());
                              if (ackedAndPrepared != null && ackedAndPrepared.contains(entry.getValue().messageId)) {
                                  continue;
                              }
                              Message msg = loadMessage(entry.getValue().location);
                              listener.recoverMessage(msg);
                          }
                      }
                  });
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          @Override
          public void recoverNextMessages(final int maxReturned, final MessageRecoveryListener listener) throws Exception {
              indexLock.writeLock().lock();
              try {
                  pageFile.tx().execute(new Transaction.Closure<Exception>() {
                      @Override
                      public void execute(Transaction tx) throws Exception {
                          StoredDestination sd = getStoredDestination(dest, tx);
                          Entry<Long, MessageKeys> entry = null;
                          int counter = recoverRolledBackAcks(destination.getPhysicalName(), sd, tx, maxReturned, listener);
                          Set<String> ackedAndPrepared = ackedAndPreparedMap.get(destination.getPhysicalName());
                          for (Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx); iterator.hasNext(); ) {
                              entry = iterator.next();
                              if (ackedAndPrepared != null && ackedAndPrepared.contains(entry.getValue().messageId)) {
                                  continue;
                              }
                              Message msg = loadMessage(entry.getValue().location);
                              msg.getMessageId().setFutureOrSequenceLong(entry.getKey());
                              listener.recoverMessage(msg);
                              counter++;
                              if (counter >= maxReturned || !listener.canRecoveryNextMessage()) {
                                  break;
                              }
                          }
                          sd.orderIndex.stoppedIterating();
                      }
                  });
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          @Override
          public void recoverMessages(final MessageRecoveryContext messageRecoveryContext) throws Exception {
  
              if(messageRecoveryContext == null || 
                 (messageRecoveryContext.getStartMessageId() != null &&
                  messageRecoveryContext.getOffset() != null)) {
                  LOG.warn("Invalid messageRecoveryContext:{}", messageRecoveryContext);
                  throw new IllegalArgumentException("Invalid messageRecoveryContext specified");
              }
  
              indexLock.writeLock().lock();
              try {
                  pageFile.tx().execute(new Transaction.Closure<Exception>() {
                      @Override
                      public void execute(Transaction tx) throws Exception {
                          StoredDestination sd = getStoredDestination(dest, tx);
  
                          /**
                           * [AMQ-9773]
                           *
                           * The order index sequence key value increases for every message since the beginning of the
                           * creation of the index, so the first message available won't be at sequence key value:0 in
                           * the index when there were older messages acknowledged.
                           *
                           * The MessageOrderCursor _position_ value is relative to the index, so there is a disconnect
                           * between queue _position_ and index sequence value over time.
                           *
                           * The MessageRecoveryContext determines the recovery start position based off the provided
                           * offset, or the position of the requested startMessageId. If a startMessageId is specified,
                           * but not found in the index, then the value of 0 is used as a fallback.
                           *
                           * The MessageRecoveryContext determines the recovery end position based off of the provided
                           * endMessageId (if provided), or the maximum recovered message count, or if the
                           * MessageRecoveryListener signals that no more messages should be recovered
                           * (ie memory is full).
                           */
                          Long startSequenceOffset = null;
                          Long endSequenceOffset = null;
  
                          if(messageRecoveryContext.getStartMessageId() != null && !messageRecoveryContext.getStartMessageId().isBlank()) {
                              startSequenceOffset = Optional.ofNullable(sd.messageIdIndex.get(tx, messageRecoveryContext.getStartMessageId())).orElse(0L);
                          } else {
                              startSequenceOffset = Optional.ofNullable(messageRecoveryContext.getOffset()).orElse(0L);
                          }
  
                          if(messageRecoveryContext.getEndMessageId() != null && !messageRecoveryContext.getEndMessageId().isBlank()) {
                              endSequenceOffset = Optional.ofNullable(sd.messageIdIndex.get(tx, messageRecoveryContext.getEndMessageId()))
                                                          .orElse(startSequenceOffset + Long.valueOf(messageRecoveryContext.getMaxMessageCountReturned()));
                              messageRecoveryContext.setEndSequenceId(endSequenceOffset);
                          }
  
                          if(endSequenceOffset != null &&
                              endSequenceOffset < startSequenceOffset) {
                              LOG.warn("Invalid offset parameters start:{} end:{}", startSequenceOffset, endSequenceOffset);
                              throw new IllegalStateException("Invalid offset parameters start:" + startSequenceOffset + " end:" + endSequenceOffset);
                          }
  
                          Entry<Long, MessageKeys> entry = null;
                          recoverRolledBackAcks(destination.getPhysicalName(), sd, tx, messageRecoveryContext.getMaxMessageCountReturned(), messageRecoveryContext);
                          Set<String> ackedAndPrepared = ackedAndPreparedMap.get(destination.getPhysicalName());
                          Iterator<Entry<Long, MessageKeys>> iterator = (messageRecoveryContext.isUseDedicatedCursor() ? sd.orderIndex.iterator(tx, new MessageOrderCursor(startSequenceOffset)) : sd.orderIndex.iterator(tx));
  
                          while (iterator.hasNext()) {
                              entry = iterator.next();
  
                              if (ackedAndPrepared != null && ackedAndPrepared.contains(entry.getValue().messageId)) {
                                  continue;
                              }
  
                              Message msg = loadMessage(entry.getValue().location);
                              msg.getMessageId().setFutureOrSequenceLong(entry.getKey());
  
                              messageRecoveryContext.recoverMessage(msg);
                              if (!messageRecoveryContext.canRecoveryNextMessage(entry.getKey())) {
                                  break;
                              }
                          }
  
                          // [AMQ-9773] The sd.orderIndex uses the destination's cursor
                          if(!messageRecoveryContext.isUseDedicatedCursor()) {
                              sd.orderIndex.stoppedIterating();
                          }
                      }
                  });
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          protected int recoverRolledBackAcks(String recoveredTxStateMapKey, StoredDestination sd, Transaction tx, int maxReturned, MessageRecoveryListener listener) throws Exception {
              int counter = 0;
              String id;
  
              Set<String> rolledBackAcks = rolledBackAcksMap.get(recoveredTxStateMapKey);
              if (rolledBackAcks == null) {
                  return counter;
              }
              for (Iterator<String> iterator = rolledBackAcks.iterator(); iterator.hasNext(); ) {
                  id = iterator.next();
                  iterator.remove();
                  Long sequence = sd.messageIdIndex.get(tx, id);
                  if (sequence != null) {
                      if (sd.orderIndex.alreadyDispatched(sequence)) {
                          listener.recoverMessage(loadMessage(sd.orderIndex.get(tx, sequence).location));
                          counter++;
                          if (counter >= maxReturned) {
                              break;
                          }
                      } else {
                          LOG.debug("rolledback ack message {} with seq {} will be picked up in future batch {}", id, sequence, sd.orderIndex.cursor);
                      }
                  } else {
                      LOG.warn("Failed to locate rolled back ack message {} in {}", id, sd);
                  }
              }
              if (rolledBackAcks.isEmpty()) {
                  rolledBackAcksMap.remove(recoveredTxStateMapKey);
              }
              return counter;
          }
  
  
          @Override
          public void resetBatching() {
              if (pageFile.isLoaded()) {
                  indexLock.writeLock().lock();
                  try {
                      pageFile.tx().execute(new Transaction.Closure<Exception>() {
                          @Override
                          public void execute(Transaction tx) throws Exception {
                              StoredDestination sd = getExistingStoredDestination(dest, tx);
                              if (sd != null) {
                                  sd.orderIndex.resetCursorPosition();}
                              }
                          });
                  } catch (Exception e) {
                      LOG.error("Failed to reset batching",e);
                  } finally {
                      indexLock.writeLock().unlock();
                  }
              }
          }
  
          @Override
          public void setBatch(final MessageId identity) throws IOException {
              indexLock.writeLock().lock();
              try {
                  pageFile.tx().execute(new Transaction.Closure<IOException>() {
                      @Override
                      public void execute(Transaction tx) throws IOException {
                          StoredDestination sd = getStoredDestination(dest, tx);
                          Long location = (Long) identity.getFutureOrSequenceLong();
                          Long pending = sd.orderIndex.minPendingAdd();
                          if (pending != null) {
                              location = Math.min(location, pending-1);
                          }
                          sd.orderIndex.setBatch(tx, location);
                      }
                  });
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          @Override
          public void setMemoryUsage(MemoryUsage memoryUsage) {
          }
          @Override
          public void start() throws Exception {
              super.start();
          }
          @Override
          public void stop() throws Exception {
              super.stop();
          }
  
          protected void lockAsyncJobQueue() {
              try {
                  if (!this.localDestinationSemaphore.tryAcquire(this.maxAsyncJobs, 60, TimeUnit.SECONDS)) {
                      throw new TimeoutException(this +" timeout waiting for localDestSem:" + this.localDestinationSemaphore);
                  }
              } catch (Exception e) {
                  LOG.error("Failed to lock async jobs for " + this.destination, e);
              }
          }
  
          protected void unlockAsyncJobQueue() {
              this.localDestinationSemaphore.release(this.maxAsyncJobs);
          }
  
          protected void acquireLocalAsyncLock() {
              try {
                  this.localDestinationSemaphore.acquire();
              } catch (InterruptedException e) {
                  LOG.error("Failed to aquire async lock for " + this.destination, e);
              }
          }
  
          protected void releaseLocalAsyncLock() {
              this.localDestinationSemaphore.release();
          }
  
          @Override
          public String toString(){
              return "permits:" + this.localDestinationSemaphore.availablePermits() + ",sd=" + storedDestinations.get(key(dest));
          }
  
          @Override
          protected void recoverMessageStoreStatistics() throws IOException {
              try {
                  MessageStoreStatistics recoveredStatistics;
                  lockAsyncJobQueue();
                  indexLock.writeLock().lock();
                  try {
                      recoveredStatistics = pageFile.tx().execute(new Transaction.CallableClosure<MessageStoreStatistics, IOException>() {
                          @Override
                          public MessageStoreStatistics execute(Transaction tx) throws IOException {
                              MessageStoreStatistics statistics = getStoredMessageStoreStatistics(dest, tx);
  
                              // Iterate through all index entries to get the size of each message
                              if (statistics == null) {
                                  StoredDestination sd = getStoredDestination(dest, tx);
                                  statistics = new MessageStoreStatistics();
                                  for (Iterator<Entry<Location, Long>> iterator = sd.locationIndex.iterator(tx); iterator.hasNext(); ) {
                                      int locationSize = iterator.next().getKey().getSize();
                                      statistics.getMessageCount().increment();
                                      statistics.getMessageSize().addSize(locationSize > 0 ? locationSize : 0);
                                  }
                              }
                              return statistics;
                          }
                      });
                      Set<String> ackedAndPrepared = ackedAndPreparedMap.get(destination.getPhysicalName());
                      if (ackedAndPrepared != null) {
                          recoveredStatistics.getMessageCount().subtract(ackedAndPrepared.size());
                      }
                      getMessageStoreStatistics().getMessageCount().setCount(recoveredStatistics.getMessageCount().getCount());
                      getMessageStoreStatistics().getMessageSize().setTotalSize(recoveredStatistics.getMessageSize().getTotalSize());
                  } finally {
                      indexLock.writeLock().unlock();
                  }
              } finally {
                  unlockAsyncJobQueue();
              }
          }
  
          @Override
          public StoreType getType() {
              return StoreType.KAHADB;
          }
      }
  
      class KahaDBTopicMessageStore extends KahaDBMessageStore implements TopicMessageStore{
          private final AtomicInteger subscriptionCount = new AtomicInteger();
          protected final MessageStoreSubscriptionStatistics messageStoreSubStats =
                  new MessageStoreSubscriptionStatistics(isEnableSubscriptionStatistics());
  
          public KahaDBTopicMessageStore(ActiveMQTopic destination) throws IOException {
              super(destination);
              this.subscriptionCount.set(getAllSubscriptions().length);
              if (isConcurrentStoreAndDispatchTopics()) {
                  asyncTopicMaps.add(asyncTaskMap);
              }
          }
  
          @Override
          protected void recoverMessageStoreStatistics() throws IOException {
              super.recoverMessageStoreStatistics();
              this.recoverMessageStoreSubMetrics();
          }
  
          @Override
          public ListenableFuture<Object> asyncAddTopicMessage(final ConnectionContext context, final Message message)
                  throws IOException {
              if (isConcurrentStoreAndDispatchTopics()) {
                  message.beforeMarshall(wireFormat);
                  StoreTopicTask result = new StoreTopicTask(this, context, message, subscriptionCount.get());
                  result.aquireLocks();
                  addTopicTask(this, result);
                  return result.getFuture();
              } else {
                  return super.asyncAddTopicMessage(context, message);
              }
          }
  
          @Override
          public void acknowledge(ConnectionContext context, String clientId, String subscriptionName,
                                  MessageId messageId, MessageAck ack) throws IOException {
              String subscriptionKey = subscriptionKey(clientId, subscriptionName).toString();
              if (isConcurrentStoreAndDispatchTopics()) {
                  AsyncJobKey key = new AsyncJobKey(messageId, getDestination());
                  StoreTopicTask task = null;
                  synchronized (asyncTaskMap) {
                      task = (StoreTopicTask) asyncTaskMap.get(key);
                  }
                  if (task != null) {
                      if (task.addSubscriptionKey(subscriptionKey)) {
                          removeTopicTask(this, messageId);
                          if (task.cancel()) {
                              synchronized (asyncTaskMap) {
                                  asyncTaskMap.remove(key);
                              }
                          }
                      }
                  } else {
                      doAcknowledge(context, subscriptionKey, messageId, ack);
                  }
              } else {
                  doAcknowledge(context, subscriptionKey, messageId, ack);
              }
          }
  
          protected void doAcknowledge(ConnectionContext context, String subscriptionKey, MessageId messageId, MessageAck ack)
                  throws IOException {
              KahaRemoveMessageCommand command = new KahaRemoveMessageCommand();
              command.setDestination(dest);
              command.setSubscriptionKey(subscriptionKey);
              command.setMessageId(messageId.toProducerKey());
              command.setTransactionInfo(ack != null ? TransactionIdConversion.convert(transactionIdTransformer.transform(ack.getTransactionId())) : null);
              if (ack != null && ack.isUnmatchedAck()) {
                  command.setAck(UNMATCHED);
              } else {
                  org.apache.activemq.util.ByteSequence packet = wireFormat.marshal(ack);
                  command.setAck(new Buffer(packet.getData(), packet.getOffset(), packet.getLength()));
              }
              store(command, false, null, null);
          }
  
          @Override
          public void addSubscription(SubscriptionInfo subscriptionInfo, boolean retroactive) throws IOException {
              String subscriptionKey = subscriptionKey(subscriptionInfo.getClientId(), subscriptionInfo
                      .getSubscriptionName());
              KahaSubscriptionCommand command = new KahaSubscriptionCommand();
              command.setDestination(dest);
              command.setSubscriptionKey(subscriptionKey.toString());
              command.setRetroactive(retroactive);
              org.apache.activemq.util.ByteSequence packet = wireFormat.marshal(subscriptionInfo);
              command.setSubscriptionInfo(new Buffer(packet.getData(), packet.getOffset(), packet.getLength()));
              store(command, isEnableJournalDiskSyncs() && true, null, null);
              this.subscriptionCount.incrementAndGet();
          }
  
          @Override
          public void deleteSubscription(String clientId, String subscriptionName) throws IOException {
              KahaSubscriptionCommand command = new KahaSubscriptionCommand();
              command.setDestination(dest);
              command.setSubscriptionKey(subscriptionKey(clientId, subscriptionName).toString());
              store(command, isEnableJournalDiskSyncs() && true, null, null);
              this.subscriptionCount.decrementAndGet();
          }
  
          @Override
          public SubscriptionInfo[] getAllSubscriptions() throws IOException {
  
              final ArrayList<SubscriptionInfo> subscriptions = new ArrayList<SubscriptionInfo>();
              indexLock.writeLock().lock();
              try {
                  pageFile.tx().execute(new Transaction.Closure<IOException>() {
                      @Override
                      public void execute(Transaction tx) throws IOException {
                          StoredDestination sd = getStoredDestination(dest, tx);
                          for (Iterator<Entry<String, KahaSubscriptionCommand>> iterator = sd.subscriptions.iterator(tx); iterator
                                  .hasNext();) {
                              Entry<String, KahaSubscriptionCommand> entry = iterator.next();
                              SubscriptionInfo info = (SubscriptionInfo) wireFormat.unmarshal(new DataInputStream(entry
                                      .getValue().getSubscriptionInfo().newInput()));
                              subscriptions.add(info);
  
                          }
                      }
                  });
              } finally {
                  indexLock.writeLock().unlock();
              }
  
              SubscriptionInfo[] rc = new SubscriptionInfo[subscriptions.size()];
              subscriptions.toArray(rc);
              return rc;
          }
  
          @Override
          public SubscriptionInfo lookupSubscription(String clientId, String subscriptionName) throws IOException {
              final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
              indexLock.writeLock().lock();
              try {
                  return pageFile.tx().execute(new Transaction.CallableClosure<SubscriptionInfo, IOException>() {
                      @Override
                      public SubscriptionInfo execute(Transaction tx) throws IOException {
                          StoredDestination sd = getStoredDestination(dest, tx);
                          KahaSubscriptionCommand command = sd.subscriptions.get(tx, subscriptionKey);
                          if (command == null) {
                              return null;
                          }
                          return (SubscriptionInfo) wireFormat.unmarshal(new DataInputStream(command
                                  .getSubscriptionInfo().newInput()));
                      }
                  });
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          @Override
          public int getMessageCount(String clientId, String subscriptionName) throws IOException {
              final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
  
              if (isEnableSubscriptionStatistics()) {
                  return (int)this.messageStoreSubStats.getMessageCount(subscriptionKey).getCount();
              } else {
  
                  indexLock.writeLock().lock();
                  try {
                      return pageFile.tx().execute(new Transaction.CallableClosure<Integer, IOException>() {
                          @Override
                          public Integer execute(Transaction tx) throws IOException {
                              StoredDestination sd = getStoredDestination(dest, tx);
                              LastAck cursorPos = getLastAck(tx, sd, subscriptionKey);
                              if (cursorPos == null) {
                                  // The subscription might not exist.
                                  return 0;
                              }
  
                              return (int) getStoredMessageCount(tx, sd, subscriptionKey);
                          }
                      });
                  } finally {
                      indexLock.writeLock().unlock();
                  }
              }
          }
  
  
          @Override
          public long getMessageSize(String clientId, String subscriptionName) throws IOException {
              final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
              if (isEnableSubscriptionStatistics()) {
                  return this.messageStoreSubStats.getMessageSize(subscriptionKey).getTotalSize();
              } else {
                  indexLock.writeLock().lock();
                  try {
                      return pageFile.tx().execute(new Transaction.CallableClosure<Long, IOException>() {
                          @Override
                          public Long execute(Transaction tx) throws IOException {
                              StoredDestination sd = getStoredDestination(dest, tx);
                              LastAck cursorPos = getLastAck(tx, sd, subscriptionKey);
                              if (cursorPos == null) {
                                  // The subscription might not exist.
                                  return 0l;
                              }
  
                              return getStoredMessageSize(tx, sd, subscriptionKey);
                          }
                      });
                  } finally {
                      indexLock.writeLock().unlock();
                  }
              }
          }
  
          protected void recoverMessageStoreSubMetrics() throws IOException {
              if (isEnableSubscriptionStatistics()) {
                  final MessageStoreSubscriptionStatistics statistics = getMessageStoreSubStatistics();
                  indexLock.writeLock().lock();
                  try {
                      pageFile.tx().execute(new Transaction.Closure<IOException>() {
                          @Override
                          public void execute(Transaction tx) throws IOException {
                              StoredDestination sd = getStoredDestination(dest, tx);
  
                              List<String> subscriptionKeys = new ArrayList<>();
                              for (Iterator<Entry<String, KahaSubscriptionCommand>> iterator = sd.subscriptions
                                      .iterator(tx); iterator.hasNext();) {
                                  Entry<String, KahaSubscriptionCommand> entry = iterator.next();
  
                                  final String subscriptionKey = entry.getKey();
                                  final LastAck cursorPos = getLastAck(tx, sd, subscriptionKey);
                                  if (cursorPos != null) {
                                      //add the subscriptions to a list for recovering pending sizes below
                                      subscriptionKeys.add(subscriptionKey);
                                      //recover just the count here as that is fast
                                      statistics.getMessageCount(subscriptionKey)
                                              .setCount(getStoredMessageCount(tx, sd, subscriptionKey));
                                  }
                              }
  
                              //Recover the message sizes for each subscription by iterating only 1 time over the order index
                              //to speed up recovery
                              final Map<String, AtomicLong> subPendingMessageSizes = getStoredMessageSize(tx, sd, subscriptionKeys);
                              subPendingMessageSizes.forEach((k,v) -> {
                                  statistics.getMessageSize(k).addSize(v.get() > 0 ? v.get() : 0);
                              });
                          }
                      });
                  } finally {
                      indexLock.writeLock().unlock();
                  }
              }
          }
  
          @Override
          public void recoverSubscription(String clientId, String subscriptionName, final MessageRecoveryListener listener)
                  throws Exception {
              final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
              @SuppressWarnings("unused")
              final SubscriptionInfo info = lookupSubscription(clientId, subscriptionName);
              indexLock.writeLock().lock();
              try {
                  pageFile.tx().execute(new Transaction.Closure<Exception>() {
                      @Override
                      public void execute(Transaction tx) throws Exception {
                          StoredDestination sd = getStoredDestination(dest, tx);
                          LastAck cursorPos = getLastAck(tx, sd, subscriptionKey);
                          SequenceSet subAckPositions = getSequenceSet(tx, sd, subscriptionKey);
                          //If we have ackPositions tracked then compare the first one as individual acknowledge mode
                          //may have bumped lastAck even though there are earlier messages to still consume
                          if (subAckPositions != null && !subAckPositions.isEmpty()
                                  && subAckPositions.getHead().getFirst() < cursorPos.lastAckedSequence) {
                              //we have messages to ack before lastAckedSequence
                              sd.orderIndex.setBatch(tx, subAckPositions.getHead().getFirst() - 1);
                          } else {
                              subAckPositions = null;
                              sd.orderIndex.setBatch(tx, cursorPos);
                          }
                          recoverRolledBackAcks(subscriptionKey, sd, tx, Integer.MAX_VALUE, listener);
                          Set<String> ackedAndPrepared = ackedAndPreparedMap.get(subscriptionKey);
                          for (Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx); iterator
                                  .hasNext();) {
                              Entry<Long, MessageKeys> entry = iterator.next();
                              if (ackedAndPrepared != null && ackedAndPrepared.contains(entry.getValue().messageId)) {
                                  continue;
                              }
                              //If subAckPositions is set then verify the sequence set contains the message still
                              //and if it doesn't skip it
                              if (subAckPositions != null && !subAckPositions.contains(entry.getKey())) {
                                  continue;
                              }
                              listener.recoverMessage(loadMessage(entry.getValue().location));
                          }
                          sd.orderIndex.resetCursorPosition();
                      }
                  });
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          @Override
          public void recoverNextMessages(String clientId, String subscriptionName, final int maxReturned,
                  final MessageRecoveryListener listener) throws Exception {
              final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
              @SuppressWarnings("unused")
              final SubscriptionInfo info = lookupSubscription(clientId, subscriptionName);
              indexLock.writeLock().lock();
              try {
                  pageFile.tx().execute(new Transaction.Closure<Exception>() {
                      @Override
                      public void execute(Transaction tx) throws Exception {
                          StoredDestination sd = getStoredDestination(dest, tx);
                          sd.orderIndex.resetCursorPosition();
                          MessageOrderCursor moc = sd.subscriptionCursors.get(subscriptionKey);
                          SequenceSet subAckPositions = getSequenceSet(tx, sd, subscriptionKey);;
                          if (moc == null) {
                              LastAck pos = getLastAck(tx, sd, subscriptionKey);
                              if (pos == null) {
                                  // sub deleted
                                  return;
                              }
                              //If we have ackPositions tracked then compare the first one as individual acknowledge mode
                              //may have bumped lastAck even though there are earlier messages to still consume
                              if (subAckPositions != null && !subAckPositions.isEmpty()
                                      && subAckPositions.getHead().getFirst() < pos.lastAckedSequence) {
                                  //we have messages to ack before lastAckedSequence
                                  sd.orderIndex.setBatch(tx, subAckPositions.getHead().getFirst() - 1);
                              } else {
                                  subAckPositions = null;
                                  sd.orderIndex.setBatch(tx, pos);
                              }
                              moc = sd.orderIndex.cursor;
                          } else {
                              sd.orderIndex.cursor.sync(moc);
                          }
  
                          Entry<Long, MessageKeys> entry = null;
                          int counter = recoverRolledBackAcks(subscriptionKey, sd, tx, maxReturned, listener);
                          Set<String> ackedAndPrepared = ackedAndPreparedMap.get(subscriptionKey);
                          for (Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx, moc); iterator
                                  .hasNext();) {
                              entry = iterator.next();
                              if (ackedAndPrepared != null && ackedAndPrepared.contains(entry.getValue().messageId)) {
                                  continue;
                              }
                              //If subAckPositions is set then verify the sequence set contains the message still
                              //and if it doesn't skip it
                              if (subAckPositions != null && !subAckPositions.contains(entry.getKey())) {
                                  continue;
                              }
                              if (listener.recoverMessage(loadMessage(entry.getValue().location))) {
                                  counter++;
                              }
                              if (counter >= maxReturned || listener.hasSpace() == false) {
                                  break;
                              }
                          }
                          sd.orderIndex.stoppedIterating();
                          if (entry != null) {
                              MessageOrderCursor copy = sd.orderIndex.cursor.copy();
                              sd.subscriptionCursors.put(subscriptionKey, copy);
                          }
                      }
                  });
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          @Override
          public Map<SubscriptionKey,List<Message>> recoverExpired(Set<SubscriptionKey> subscriptions, int maxBrowse,
              MessageRecoveryListener listener) throws Exception {
              indexLock.writeLock().lock();
              try {
                  return pageFile.tx().execute(tx -> {
                          StoredDestination sd = getStoredDestination(dest, tx);
                          sd.orderIndex.resetCursorPosition();
                          int count = 0;
                          final Map<SubscriptionKey, List<Message>> expired = new HashMap<>();
                          final Map<String, SubscriptionKey> subKeys = new HashMap<>();
  
                          // Check each subscription and track the ones that exist
                          for (SubscriptionKey sub : subscriptions) {
                              final String subKeyString = subscriptionKey(sub.getClientId(), sub.getSubscriptionName());
                              if (sd.subscriptionCache.contains(subKeyString)) {
                                  subKeys.put(subKeyString, sub);
                              }
                          }
  
                          // Iterate one time through the topic and check each message, stopping if we run out
                          // hit the max browse limit, or if the listener returns false for hasSpace()
                          final Set<Long> uniqueExpired = new HashSet<>();
                          for (Iterator<Entry<Long, MessageKeys>> iterator =
                              sd.orderIndex.iterator(tx, new MessageOrderCursor()); count < maxBrowse && iterator.hasNext() && listener.hasSpace(); ) {
                              count++;
                              Entry<Long, MessageKeys> entry = iterator.next();
                              Set<String> ackedAndPrepared = ackedAndPreparedMap.get(destination.getPhysicalName());
                              if (ackedAndPrepared != null && ackedAndPrepared.contains(entry.getValue().messageId)) {
                                  continue;
                              }
  
                              final Message msg = loadMessage(entry.getValue().location);
                              if (msg.isExpired()) {
                                  // For every message that is expired, go through and check each subscription to see
                                  // if the message has already been acked. We don't want to return subs that have already
                                  // acked the message.
                                  for(Entry<String, SubscriptionKey> subKeyEntry : subKeys.entrySet()) {
                                      SequenceSet sequence = sd.ackPositions.get(tx, subKeyEntry.getKey());
                                      if (sequence != null && sequence.contains(entry.getKey())) {
                                          List<Message> expMessages = expired.computeIfAbsent(subKeyEntry.getValue(), m -> new ArrayList<>());
                                          expMessages.add(msg);
                                          // pass unique messages to the listener so it can do any custom
                                          // handling for the next call to listener.hasSpace()
                                          if (uniqueExpired.add(entry.getKey())) {
                                              listener.recoverMessage(msg);
                                          }
                                      }
                                  }
                              }
                          }
                          return expired;
                      });
              } finally {
                  indexLock.writeLock().unlock();
              }
          }
  
          @Override
          public void resetBatching(String clientId, String subscriptionName) {
              try {
                  final String subscriptionKey = subscriptionKey(clientId, subscriptionName);
                  indexLock.writeLock().lock();
                  try {
                      pageFile.tx().execute(new Transaction.Closure<IOException>() {
                          @Override
                          public void execute(Transaction tx) throws IOException {
                              StoredDestination sd = getStoredDestination(dest, tx);
                              sd.subscriptionCursors.remove(subscriptionKey);
                          }
                      });
                  }finally {
                      indexLock.writeLock().unlock();
                  }
              } catch (IOException e) {
                  throw new RuntimeException(e);
              }
          }
  
          @Override
          public MessageStoreSubscriptionStatistics getMessageStoreSubStatistics() {
              return messageStoreSubStats;
          }
      }
  
      String subscriptionKey(String clientId, String subscriptionName) {
          return clientId + ":" + subscriptionName;
      }
  
      @Override
      public MessageStore createQueueMessageStore(ActiveMQQueue destination) throws IOException {
          String key = key(convert(destination));
          MessageStore store = storeCache.get(key(convert(destination)));
          if (store == null) {
              final MessageStore queueStore = this.transactionStore.proxy(new KahaDBMessageStore(destination));
              store = storeCache.putIfAbsent(key, queueStore);
              if (store == null) {
                  store = queueStore;
              }
          }
  
          return store;
      }
  
      @Override
      public TopicMessageStore createTopicMessageStore(ActiveMQTopic destination) throws IOException {
          String key = key(convert(destination));
          MessageStore store = storeCache.get(key(convert(destination)));
          if (store == null) {
              final TopicMessageStore topicStore = this.transactionStore.proxy(new KahaDBTopicMessageStore(destination));
              store = storeCache.putIfAbsent(key, topicStore);
              if (store == null) {
                  store = topicStore;
              }
          }
  
          return (TopicMessageStore) store;
      }
  
      /**
       * Cleanup method to remove any state associated with the given destination.
       * This method does not stop the message store (it might not be cached).
       *
       * @param destination
       *            Destination to forget
       */
      @Override
      public void removeQueueMessageStore(ActiveMQQueue destination) {
      }
  
      /**
       * Cleanup method to remove any state associated with the given destination
       * This method does not stop the message store (it might not be cached).
       *
       * @param destination
       *            Destination to forget
       */
      @Override
      public void removeTopicMessageStore(ActiveMQTopic destination) {
      }
  
      @Override
      public void deleteAllMessages() throws IOException {
          deleteAllMessages = true;
      }
  
      @Override
      public Set<ActiveMQDestination> getDestinations() {
          try {
              final HashSet<ActiveMQDestination> rc = new HashSet<ActiveMQDestination>();
              indexLock.writeLock().lock();
              try {
                  pageFile.tx().execute(new Transaction.Closure<IOException>() {
                      @Override
                      public void execute(Transaction tx) throws IOException {
                          for (Iterator<Entry<String, StoredDestination>> iterator = metadata.destinations.iterator(tx); iterator
                                  .hasNext();) {
                              Entry<String, StoredDestination> entry = iterator.next();
                              //Removing isEmpty topic check - see AMQ-5875
                              rc.add(convert(entry.getKey()));
                          }
                      }
                  });
              }finally {
                  indexLock.writeLock().unlock();
              }
              return rc;
          } catch (IOException e) {
              throw new RuntimeException(e);
          }
      }
  
      @Override
      public long getLastMessageBrokerSequenceId() throws IOException {
          return 0;
      }
  
      @Override
      public long getLastProducerSequenceId(ProducerId id) {
          indexLock.writeLock().lock();
          try {
              return metadata.producerSequenceIdTracker.getLastSeqId(id);
          } finally {
              indexLock.writeLock().unlock();
          }
      }
  
      @Override
      public long size() {
          try {
              return journalSize.get() + getPageFile().getDiskSize();
          } catch (IOException e) {
              throw new RuntimeException(e);
          }
      }
  
      @Override
      public void beginTransaction(ConnectionContext context) throws IOException {
          throw new IOException("Not yet implemented.");
      }
      @Override
      public void commitTransaction(ConnectionContext context) throws IOException {
          throw new IOException("Not yet implemented.");
      }
      @Override
      public void rollbackTransaction(ConnectionContext context) throws IOException {
          throw new IOException("Not yet implemented.");
      }
  
      @Override
      public void checkpoint(boolean sync) throws IOException {
          super.checkpointCleanup(sync);
      }
  
      // /////////////////////////////////////////////////////////////////
      // Internal helper methods.
      // /////////////////////////////////////////////////////////////////
  
      /**
       * @param location
       * @return
       * @throws IOException
       */
      Message loadMessage(Location location) throws IOException {
          try {
              JournalCommand<?> command = load(location);
              KahaAddMessageCommand addMessage = null;
              switch (command.type()) {
                  case KAHA_UPDATE_MESSAGE_COMMAND:
                      addMessage = ((KahaUpdateMessageCommand) command).getMessage();
                      break;
                  case KAHA_ADD_MESSAGE_COMMAND:
                      addMessage = (KahaAddMessageCommand) command;
                      break;
                  default:
                      throw new IOException("Could not load journal record, unexpected command type: " + command.type() + " at location: " + location);
              }
              if (!addMessage.hasMessage()) {
                  throw new IOException("Could not load journal record, null message content at location: " + location);
              }
              Message msg = (Message) wireFormat.unmarshal(new DataInputStream(addMessage.getMessage().newInput()));
              return msg;
          } catch (Throwable t) {
              IOException ioe = IOExceptionSupport.create("Unexpected error on journal read at: " + location , t);
              LOG.error("Failed to load message at: {}", location , ioe);
              brokerService.handleIOException(ioe);
              throw ioe;
          }
      }
  
      // /////////////////////////////////////////////////////////////////
      // Internal conversion methods.
      // /////////////////////////////////////////////////////////////////
  
      KahaLocation convert(Location location) {
          KahaLocation rc = new KahaLocation();
          rc.setLogId(location.getDataFileId());
          rc.setOffset(location.getOffset());
          return rc;
      }
  
      KahaDestination convert(ActiveMQDestination dest) {
          KahaDestination rc = new KahaDestination();
          rc.setName(dest.getPhysicalName());
          switch (dest.getDestinationType()) {
          case ActiveMQDestination.QUEUE_TYPE:
              rc.setType(DestinationType.QUEUE);
              return rc;
          case ActiveMQDestination.TOPIC_TYPE:
              rc.setType(DestinationType.TOPIC);
              return rc;
          case ActiveMQDestination.TEMP_QUEUE_TYPE:
              rc.setType(DestinationType.TEMP_QUEUE);
              return rc;
          case ActiveMQDestination.TEMP_TOPIC_TYPE:
              rc.setType(DestinationType.TEMP_TOPIC);
              return rc;
          default:
              return null;
          }
      }
  
      ActiveMQDestination convert(String dest) {
          int p = dest.indexOf(":");
          if (p < 0) {
              throw new IllegalArgumentException("Not in the valid destination format");
          }
          int type = Integer.parseInt(dest.substring(0, p));
          String name = dest.substring(p + 1);
          return convert(type, name);
      }
  
      private ActiveMQDestination convert(KahaDestination commandDestination) {
          return convert(commandDestination.getType().getNumber(), commandDestination.getName());
      }
  
      private ActiveMQDestination convert(int type, String name) {
          switch (KahaDestination.DestinationType.valueOf(type)) {
          case QUEUE:
              return new ActiveMQQueue(name);
          case TOPIC:
              return new ActiveMQTopic(name);
          case TEMP_QUEUE:
              return new ActiveMQTempQueue(name);
          case TEMP_TOPIC:
              return new ActiveMQTempTopic(name);
          default:
              throw new IllegalArgumentException("Not in the valid destination format");
          }
      }
  
      public TransactionIdTransformer getTransactionIdTransformer() {
          return transactionIdTransformer;
      }
  
      public void setTransactionIdTransformer(TransactionIdTransformer transactionIdTransformer) {
          this.transactionIdTransformer = transactionIdTransformer;
      }
  
      static class AsyncJobKey {
          MessageId id;
          ActiveMQDestination destination;
  
          AsyncJobKey(MessageId id, ActiveMQDestination destination) {
              this.id = id;
              this.destination = destination;
          }
  
          @Override
          public boolean equals(Object obj) {
              if (obj == this) {
                  return true;
              }
              return obj instanceof AsyncJobKey && id.equals(((AsyncJobKey) obj).id)
                      && destination.equals(((AsyncJobKey) obj).destination);
          }
  
          @Override
          public int hashCode() {
              return id.hashCode() + destination.hashCode();
          }
  
          @Override
          public String toString() {
              return destination.getPhysicalName() + "-" + id;
          }
      }
  
      public interface StoreTask {
          public boolean cancel();
  
          public void aquireLocks();
  
          public void releaseLocks();
      }
  
      class StoreQueueTask implements Runnable, StoreTask {
          protected final Message message;
          protected final ConnectionContext context;
          protected final KahaDBMessageStore store;
          protected final InnerFutureTask future;
          protected final AtomicBoolean done = new AtomicBoolean();
          protected final AtomicBoolean locked = new AtomicBoolean();
  
          public StoreQueueTask(KahaDBMessageStore store, ConnectionContext context, Message message) {
              this.store = store;
              this.context = context;
              this.message = message;
              this.future = new InnerFutureTask(this);
          }
  
          public ListenableFuture<Object> getFuture() {
              return this.future;
          }
  
          @Override
          public boolean cancel() {
              if (this.done.compareAndSet(false, true)) {
                  return this.future.cancel(false);
              }
              return false;
          }
  
          @Override
          public void aquireLocks() {
              if (this.locked.compareAndSet(false, true)) {
                  try {
                      globalQueueSemaphore.acquire();
                      store.acquireLocalAsyncLock();
                      message.incrementReferenceCount();
                  } catch (InterruptedException e) {
                      LOG.warn("Failed to aquire lock", e);
                  }
              }
  
          }
  
          @Override
          public void releaseLocks() {
              if (this.locked.compareAndSet(true, false)) {
                  store.releaseLocalAsyncLock();
                  globalQueueSemaphore.release();
                  message.decrementReferenceCount();
              }
          }
  
          @Override
          public void run() {
              this.store.doneTasks++;
              try {
                  if (this.done.compareAndSet(false, true)) {
                      this.store.addMessage(context, message);
                      removeQueueTask(this.store, this.message.getMessageId());
                      this.future.complete();
                  } else if (cancelledTaskModMetric > 0 && (++this.store.canceledTasks) % cancelledTaskModMetric == 0) {
                      System.err.println(this.store.dest.getName() + " cancelled: "
                              + (this.store.canceledTasks / this.store.doneTasks) * 100);
                      this.store.canceledTasks = this.store.doneTasks = 0;
                  }
              } catch (Throwable t) {
                  this.future.setException(t);
                  removeQueueTask(this.store, this.message.getMessageId());
              }
          }
  
          protected Message getMessage() {
              return this.message;
          }
  
          private class InnerFutureTask extends FutureTask<Object> implements ListenableFuture<Object>  {
  
              private final AtomicReference<Runnable> listenerRef = new AtomicReference<>();
  
              public InnerFutureTask(Runnable runnable) {
                  super(runnable, null);
              }
  
              @Override
              public void setException(final Throwable e) {
                  super.setException(e);
              }
  
              public void complete() {
                  super.set(null);
              }
  
              @Override
              public void done() {
                  fireListener();
              }
  
              @Override
              public void addListener(Runnable listener) {
                  this.listenerRef.set(listener);
                  if (isDone()) {
                      fireListener();
                  }
              }
  
              private void fireListener() {
                  Runnable listener = listenerRef.getAndSet(null);
                  if (listener != null) {
                      try {
                          listener.run();
                      } catch (Exception ignored) {
                          LOG.warn("Unexpected exception from future {} listener callback {}", this, listener, ignored);
                      }
                  }
              }
          }
      }
  
      class StoreTopicTask extends StoreQueueTask {
          private final int subscriptionCount;
          private final List<String> subscriptionKeys = new ArrayList<String>(1);
          private final KahaDBTopicMessageStore topicStore;
          public StoreTopicTask(KahaDBTopicMessageStore store, ConnectionContext context, Message message,
                  int subscriptionCount) {
              super(store, context, message);
              this.topicStore = store;
              this.subscriptionCount = subscriptionCount;
  
          }
  
          @Override
          public void aquireLocks() {
              if (this.locked.compareAndSet(false, true)) {
                  try {
                      globalTopicSemaphore.acquire();
                      store.acquireLocalAsyncLock();
                      message.incrementReferenceCount();
                  } catch (InterruptedException e) {
                      LOG.warn("Failed to aquire lock", e);
                  }
              }
          }
  
          @Override
          public void releaseLocks() {
              if (this.locked.compareAndSet(true, false)) {
                  message.decrementReferenceCount();
                  store.releaseLocalAsyncLock();
                  globalTopicSemaphore.release();
              }
          }
  
          /**
           * add a key
           *
           * @param key
           * @return true if all acknowledgements received
           */
          public boolean addSubscriptionKey(String key) {
              synchronized (this.subscriptionKeys) {
                  this.subscriptionKeys.add(key);
              }
              return this.subscriptionKeys.size() >= this.subscriptionCount;
          }
  
          @Override
          public void run() {
              this.store.doneTasks++;
              try {
                  if (this.done.compareAndSet(false, true)) {
                      this.topicStore.addMessage(context, message);
                      // apply any acks we have
                      synchronized (this.subscriptionKeys) {
                          for (String key : this.subscriptionKeys) {
                              this.topicStore.doAcknowledge(context, key, this.message.getMessageId(), null);
  
                          }
                      }
                      removeTopicTask(this.topicStore, this.message.getMessageId());
                      this.future.complete();
                  } else if (cancelledTaskModMetric > 0 && this.store.canceledTasks++ % cancelledTaskModMetric == 0) {
                      System.err.println(this.store.dest.getName() + " cancelled: "
                              + (this.store.canceledTasks / this.store.doneTasks) * 100);
                      this.store.canceledTasks = this.store.doneTasks = 0;
                  }
              } catch (Throwable t) {
                  this.future.setException(t);
                  removeTopicTask(this.topicStore, this.message.getMessageId());
              }
          }
      }
  
      public class StoreTaskExecutor extends ThreadPoolExecutor {
  
          public StoreTaskExecutor(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit timeUnit, BlockingQueue<Runnable> queue, ThreadFactory threadFactory) {
              super(corePoolSize, maximumPoolSize, keepAliveTime, timeUnit, queue, threadFactory);
          }
  
          @Override
          protected void afterExecute(Runnable runnable, Throwable throwable) {
              super.afterExecute(runnable, throwable);
  
              if (runnable instanceof StoreTask) {
                 ((StoreTask)runnable).releaseLocks();
              }
          }
      }
  
      @Override
      public JobSchedulerStore createJobSchedulerStore() throws IOException, UnsupportedOperationException {
          return new JobSchedulerStoreImpl();
      }
  
      /* (non-Javadoc)
       * @see org.apache.activemq.store.NoLocalSubscriptionAware#isPersistNoLocal()
       */
      @Override
      public boolean isPersistNoLocal() {
          // Prior to v11 the broker did not store the noLocal value for durable subs.
          return brokerService.getStoreOpenWireVersion() >= 11;
      }
  }
  