{@inheritDoc}

@throws QueueException.EphemeralIsFull
if the ephemeral storage is full
558559
@Override
    public IQueueMessage<ID, DATA> take() throws QueueException.EphemeralIsFull {
        if (!isEphemeralDisabled()) {
            int ephemeralMaxSize = getEphemeralMaxSize();
            if (ephemeralMaxSize > 0 && ephemeralSize() >= ephemeralMaxSize) {
                throw new QueueException.EphemeralIsFull(ephemeralMaxSize);
            }
        }
        lockTake.lock();
        try {
            if (lastFetchedId == null) {
                itQueue.seekToFirst();
            } else {
                itQueue.seek(lastFetchedId);
            }
            if (!itQueue.isValid()) {
                return null;
            }
            lastFetchedId = itQueue.key();
            byte[] value = itQueue.value();
            IQueueMessage<ID, DATA> msg = deserialize(value);
            try {
                batchTake.delete(cfQueue, lastFetchedId);
                // batchTake.put(cfMetadata, keyLastFetchedId, lastFetchedId);
                batchTake.put(cfMetadata, keyLastFetchedIdBytes, lastFetchedId);
                if (!isEphemeralDisabled() && msg != null) {
                    byte[] _key = msg.getId().toString().getBytes(QueueUtils.UTF8);
                    batchTake.put(cfEphemeral, _key, value);
                }
                rocksDbWrapper.write(writeOptions, batchTake);
            } catch (RocksDBException e) {
                throw new QueueException(e);
            } finally {
                batchTake.clear();
            }
            itQueue.next();
            return msg;
        } finally {
            lockTake.unlock();
        }
    }
918193
private void checkStorageState(StorageException e) {
        if (e.getCause() instanceof AmazonS3Exception) {
            String errorCode =
                ((AmazonS3Exception) e.getCause()).getErrorCode();
            if (INVALID_OBJECT_STATE.equals(errorCode)) {
                String message = "The storage state of this content item " +
                                 "does not allow for this action to be taken. To resolve " +
                                 "this issue: 1. Request that this content item be " +
                                 "retrieved from offline storage 2. Wait (retrieval may " +
                                 "take up to 5 hours) 3. Retry this request";
                throw new StorageStateException(message, e);
            }
        }
    }
12358
private void queueRequest(ManagedConnectionFactory managedConnectionFactory, long waitTimeout) throws ResourceAllocationException, ConnectionWaitTimeoutException {

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.entry(this, tc, "queueRequest", waitTimeout);
        }
        if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled())
            Tr.debug(tc, "Current connection pool", pm.toString());
        if (waitTimeout == 0) {
            --pm.waiterCount;
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                Tr.debug(
                         tc,
                         "Timeout.  Decremented waiterCount, which is now "
                             + pm.waiterCount
                             + " on datasource "
                             + gConfigProps.cfName);
            }
            Object[] connWaitTimeoutparms = new Object[] { "queueRequest", gConfigProps.cfName, 0, pm.waiterCount,
                                                           pm.totalConnectionCount.get() };
            Tr.error(
                     tc,
                     "POOL_MANAGER_EXCP_CCF2_0001_J2CA0045",
                     connWaitTimeoutparms);
            String connWaitTimeoutMessage = Tr.formatMessage(tc, "POOL_MANAGER_EXCP_CCF2_0001_J2CA0045", connWaitTimeoutparms);
            ConnectionWaitTimeoutException cwte = new ConnectionWaitTimeoutException(connWaitTimeoutMessage);
            com.ibm.ws.ffdc.FFDCFilter.processException(cwte, J2CConstants.DMSID_MAX_CONNECTIONS_REACHED, "192", this.pm);
            pm.activeRequest.decrementAndGet();

            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                Tr.exit(this, tc, "queueRequest", cwte);
            throw cwte;
        }
        if ((tc.isDebugEnabled())) {
            ++freePoolQueuedRequests;
            if (TraceComponent.isAnyTracingEnabled() && tc.isDebugEnabled()) {
                String poolStats = pm.gatherPoolStatisticalData();
                Tr.debug(this, tc, "Queueing Waiter for pool <" + gConfigProps.getXpathId() + ">. Current Pool Stats are:");
                Tr.debug(this, tc, poolStats);
            }
        }
        try {
            if (pm.displayInfiniteWaitMessage) {
                Tr.info(tc, "INFINITE_CONNECTION_WAIT_TIMEOUT_J2CA0127", gConfigProps.getXpathId());

                pm.displayInfiniteWaitMessage = false; // only display this message once per PM.
            }
            pm.activeRequest.decrementAndGet();
            if (waitTimeout < 0) {
                pm.waiterFreePoolLock.wait(0); //wait an infinite amount of time
            } else {
                pm.waiterFreePoolLock.wait(waitTimeout); //wait the specified amount of time
            }
            pm.requestingAccessToPool();
        } catch (InterruptedException ie) {
            if (tc.isDebugEnabled()) {
                Tr.debug(this, tc, "Interupted waiting for a connection");
            }

            synchronized (pm.waiterFreePoolLock) {
                --pm.waiterCount;
            }

            if (tc.isDebugEnabled()) {
                if (pm.waiterCount == 0) {
                    pm.waitersEndedTime = System.currentTimeMillis();
                    Tr.debug(this, tc, "Waiters: requests for connections are no longer being queued. End Time:" + pm.waitersEndedTime);
                    Tr.debug(this, tc, "Waiters: total time waiter were in queue: " + (pm.waitersEndedTime - pm.waitersStartedTime));
                }
            }
            ResourceAllocationException throwMe = new ResourceAllocationException(ie.getMessage());
            throwMe.initCause(ie);
            if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled())
                Tr.exit(this, tc, "queueRequest", throwMe);
            throw throwMe;
        }

        if (TraceComponent.isAnyTracingEnabled() && tc.isEntryEnabled()) {
            Tr.exit(this, tc, "queueRequest");
        }
    }
769504
@Override
    public final boolean retainAll(Collection<?> c) {
        this.writeLock.lock();
        try {
            int newSize = 0;
            for (final Queue<T> queue : this.keyedQueues.values()) {
                queue.retainAll(c);
                newSize += queue.size();
            }

            // Update the queue size
            final int oldSize = this.size;
            this.size = newSize;

            // If the updated size and old size differ things changed
            return this.size != oldSize;
        } finally {
            this.writeLock.unlock();
        }
    }
1247579
public void enqueueAll(
      final Collection<Pair<ExecutionReference, ExecutableFlow>> collection)
      throws ExecutorManagerException {
    for (final Pair<ExecutionReference, ExecutableFlow> pair : collection) {
      enqueue(pair.getSecond(), pair.getFirst());
    }
  }
200168
public void queueEntry(String entryPath, InputStream inStream,
                           int maxBytes)
                           throws IOException, TarMalformatException {
        entryQueue.add(new TarEntrySupplicant(entryPath, inStream, maxBytes,
                                              '0', archive));
    }
487264
public void flush() throws JournalException
    {
    	if (failing)
    		throw new JournalException("Store journal is failing");
    	
    	boolean newFlushRequired = false;
    	synchronized (journalWriteQueue)
        {
    		if (unflushedJournalSize > maxUnflushedJournalSize)
    		{
    			newFlushRequired = !flushingJournal;
    			if (newFlushRequired)
    				flushingJournal = true;
    			unflushedJournalSize = 0;
    		}
        }
    	
    	// Flush journal to store (asynchronous)
    	if (newFlushRequired)
        {
        	try
        	{
        		asyncTaskManager.execute(flushJournalAsyncTask);
        	}
        	catch (JMSException e)
        	{
        		throw new JournalException("Cannot flush journal asynchronously : "+e);
        	}
        }
    }
756298
private void purgeQueue(Queue queue, Session session) throws JMSException {
        purgeDestination(queue, session, queue.getQueueName());
    }
292858
@Override
	protected ThreadBase<E> getSpillingThread(ExceptionHandler<IOException> exceptionHandler, CircularQueues<E> queues,
			AbstractInvokable parentTask, MemoryManager memoryManager, IOManager ioManager, 
			TypeSerializerFactory<E> serializerFactory, TypeComparator<E> comparator,
			List<MemorySegment> sortReadMemory, List<MemorySegment> writeMemory, int maxFileHandles)
	{
		this.bufferIterator = new BufferQueueIterator(queues);
		setResultIterator(this.bufferIterator);
		
		return null;
	}
814275
public function clear()
    {
        $cleared = false;
        try {
            /* @todo Not at all sure about this... do we really want to flush()? What happens if the storage supports
            /*       namespaces, but none has been provided? */
            $options = $this->storage->getOptions();
            $namespace = $options->getNamespace();
            if ($this->storage instanceof ClearByNamespaceInterface && $namespace) {
                $cleared = $this->storage->clearByNamespace($namespace);
            } elseif ($this->storage instanceof FlushableInterface) {
                $cleared = $this->storage->flush();
            } else {
                throw new CacheException(sprintf("Storage %s does not support clear()", get_class($this->storage)));
            }
        } catch (Exception\InvalidArgumentException $e) {
            throw new InvalidArgumentException($e->getMessage(), $e->getCode(), $e);

        } catch (Exception\ExceptionInterface $e) {
            throw new CacheException($e->getMessage(), $e->getCode(), $e);
        }

        return $cleared;
    }
1216705
@Override
  public void deleteObjects(List<StorageResourceId> fullObjectNames) throws IOException {
    logger.atFine().log("deleteObjects(%s)", fullObjectNames);

    if (fullObjectNames.isEmpty()) {
      return;
    }

    // Validate that all the elements represent StorageObjects.
    for (StorageResourceId fullObjectName : fullObjectNames) {
      Preconditions.checkArgument(
          fullObjectName.isStorageObject(),
          "Expected full StorageObject names only, got: %s", fullObjectName);
    }

    // Gather exceptions to wrap in a composite exception at the end.
    final KeySetView<IOException, Boolean> innerExceptions = ConcurrentHashMap.newKeySet();
    BatchHelper batchHelper =
        batchFactory.newBatchHelper(
            httpRequestInitializer,
            gcs,
            storageOptions.getMaxRequestsPerBatch(),
            fullObjectNames.size(),
            storageOptions.getBatchThreads());

    for (StorageResourceId fullObjectName : fullObjectNames) {
      queueSingleObjectDelete(fullObjectName, innerExceptions, batchHelper, 1);
    }

    batchHelper.flush();

    if (innerExceptions.size() > 0) {
      throw GoogleCloudStorageExceptions.createCompositeException(innerExceptions);
    }
  }
147634
protected function getQueue()
        {
            if (null === $this->queue) {
                $this->queue = new $this->queueClass();

                if (!$this->queue instanceof SplPriorityQueue) {
                    throw new Exception(sprintf(
                        'Queue expects an internal queue of type SplPriorityQueue; received "%s"',
                        get_class($this->queue)
                    ));
                }
            }

            return $this->queue;
        }
1041183
private synchronized boolean isQueueEmptyForSessionId (final short sessionId) {
     if (tc.isEntryEnabled()) SibTr.entry(this, tc, "isQueueEmptyForSessionId", "sessionId="+sessionId);

     boolean empty = true;
     final Iterator it = queue.iterator();
     while(it.hasNext()) {
       final QueueData data = (QueueData)it.next();
       final ProxyQueue pq = data.getProxyQueue();
       if (pq.getId() == sessionId) {
         if (tc.isDebugEnabled()) SibTr.debug(this, tc, "Found match: ", pq);
         empty = false;
         break;
       }
     }

     if (tc.isEntryEnabled()) SibTr.exit(this, tc, "isQueueEmptyForSessionId", Boolean.valueOf(empty));
     return empty;
   }
137391
public synchronized void enqueue(final T o) throws IOException {
		assert o != null;
		fbaos.reset();
		BinIO.storeObject(o, fbaos);
		byteDiskQueue.enqueueInt(fbaos.length);
		byteDiskQueue.enqueue(fbaos.array, 0, fbaos.length);
		size++;
	}
925663
public synchronized void enqueue(final byte[] array) throws IOException {
		assert array != null;
		byteDiskQueue.enqueueInt(array.length);
		byteDiskQueue.enqueue(array);
		size++;
	}
336986
final void waitForSizeQueue(final int queueSize) {
    synchronized (this.queue) {
      while (this.queue.size() > queueSize) {
        try {
          this.queue.wait(250L);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
        }
      }
      try {
        Thread.sleep(500L);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
      this.queue.notifyAll();
    }
  }
794057
@Override
  public Page allocate(int size, boolean thief, boolean victim, OffHeapStorageArea owner) {
      while (true) {
          processQueue();
          long now = max.get();
          if (now < size) {
            return null;
          } else if (max.compareAndSet(now, now - size)) {
            ByteBuffer buffer;
            try {
              buffer = ByteBuffer.allocateDirect(size);
            } catch (OutOfMemoryError e) {
              return null;
            }
            bufferSizes.put(new PhantomReference<>(buffer, allocatedBuffers), size);
            return new Page(buffer, owner);
          }
      }
  }