docs: @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();
          }
      }