docs: private ListenableFuture<?> write(final Callable<BBContainer> tupleDataC, final boolean prependLength) {
          /*
           * Unwrap the data to be written. For the traditional
           * snapshot data target this should be a noop.
           */
          BBContainer tupleDataTemp;
          try {
              tupleDataTemp = tupleDataC.call();
              /*
               * Can be null if the dedupe filter nulled out the buffer
               */
              if (tupleDataTemp == null) {
                  return Futures.immediateFuture(null);
              }
          } catch (Throwable t) {
              return Futures.immediateFailedFuture(t);
          }
          final BBContainer tupleDataCont = tupleDataTemp;
  
  
          if (m_writeFailed) {
              tupleDataCont.discard();
              return null;
          }
  
          ByteBuffer tupleData = tupleDataCont.b();
  
          m_outstandingWriteTasks.incrementAndGet();
  
          Future<BBContainer> compressionTask = null;
          if (prependLength) {
              BBContainer cont =
                      DBBPool.allocateDirectAndPool(SnapshotSiteProcessor.m_snapshotBufferCompressedLen);
              //Skip 4-bytes so the partition ID is not compressed
              //That way if we detect a corruption we know what partition is bad
              tupleData.position(tupleData.position() + 4);
              /*
               * Leave 12 bytes, it's going to be a 4-byte length prefix, a 4-byte partition id,
               * and a 4-byte CRC32C of just the header bytes, in addition to the compressed payload CRC
               * that is 16 bytes, but 4 of those are done by CompressionService
               */
              cont.b().position(12);
              compressionTask = CompressionService.compressAndCRC32cBufferAsync(tupleData, cont);
          }
          final Future<BBContainer> compressionTaskFinal = compressionTask;
  
          ListenableFuture<?> writeTask = m_es.submit(new Callable<Object>() {
              @Override
              public Object call() throws Exception {
                  int permitAcquired = 0;
                  try {
                      if (m_acceptOneWrite) {
                          m_acceptOneWrite = false;
                      } else {
                          if (m_simulateBlockedWrite != null) {
                              m_simulateBlockedWrite.await();
                          }
                          if (m_simulateFullDiskWritingChunk) {
                              //Make sure to consume the result of the compression
                              compressionTaskFinal.get().discard();
                              throw new IOException("Disk full");
                          }
                      }
  
                      final ByteBuffer tupleData = tupleDataCont.b();
                      int totalWritten = 0;
                      if (prependLength) {
                          BBContainer payloadContainer = compressionTaskFinal.get();
                          try {
                              final ByteBuffer payloadBuffer = payloadContainer.b();
                              payloadBuffer.position(0);
  
                              ByteBuffer lengthPrefix = ByteBuffer.allocate(12);
                              permitAcquired = payloadBuffer.remaining();
                              m_bytesAllowedBeforeSync.acquire(permitAcquired);
                              //Length prefix does not include 4 header items, just compressd payload
                              //that follows
                              lengthPrefix.putInt(payloadBuffer.remaining() - 16);//length prefix
                              lengthPrefix.putInt(tupleData.getInt(0)); // partitionId
  
                              /*
                               * Checksum the header and put it in the payload buffer
                               */
                              PureJavaCrc32C crc = new PureJavaCrc32C();
                              crc.update(lengthPrefix.array(), 0, 8);
                              lengthPrefix.putInt((int)crc.getValue());
                              lengthPrefix.flip();
                              payloadBuffer.put(lengthPrefix);
                              payloadBuffer.position(0);
  
                              enforceSnapshotRateLimit(payloadBuffer.remaining());
  
                              /*
                               * Write payload to file
                               */
                              while (payloadBuffer.hasRemaining()) {
                                  totalWritten += m_channel.write(payloadBuffer);
                              }
                          } finally {
                              payloadContainer.discard();
                          }
                      } else {
                          permitAcquired = tupleData.remaining();
                          m_bytesAllowedBeforeSync.acquire(permitAcquired);
                          while (tupleData.hasRemaining()) {
                              totalWritten += m_channel.write(tupleData);
                          }
                      }
                      m_bytesWritten += totalWritten;
                      m_bytesWrittenSinceLastSync.addAndGet(totalWritten);
                  } catch (IOException e) {
                      if (permitAcquired > 0) {
                          m_bytesAllowedBeforeSync.release(permitAcquired);
                      }
                      m_writeException = e;
                      SNAP_LOG.error("Error while attempting to write snapshot data to file " + m_file, e);
                      m_writeFailed = true;
                      throw e;
                  } finally {
                      try {
                          tupleDataCont.discard();
                      } finally {
                          m_outstandingWriteTasksLock.lock();
                          try {
                              if (m_outstandingWriteTasks.decrementAndGet() == 0) {
                                  m_noMoreOutstandingWriteTasksCondition.signalAll();
                              }
                          } finally {
                              m_outstandingWriteTasksLock.unlock();
                          }
                      }
                  }
                  return null;
              }
          });
          return writeTask;
      }