docs: private RestoreWork processMessage(DecodedContainer msg,
                                         CachedByteBufferAllocator resultBufferAllocator) {
          if (msg == null) {
              return null;
          }
  
          RestoreWork restoreWork = null;
          try {
              if (msg.m_msgType == StreamSnapshotMessageType.FAILURE) {
                  VoltDB.crashLocalVoltDB("Rejoin source sent failure message.", false, null);
  
                  // for test code only
                  if (m_expectedEOFs.decrementAndGet() == 0) {
                      m_EOF = true;
                  }
              }
              else if (msg.m_msgType == StreamSnapshotMessageType.END) {
                  if (rejoinLog.isTraceEnabled()) {
                      rejoinLog.trace("Got END message " + msg.m_blockIndex + " from " +
                              CoreUtils.hsIdToString(msg.m_srcHSId) +
                              " (TargetId " + msg.m_dataTargetId + ")");
                  }
  
                  if (m_expectedEOFs.decrementAndGet() == 0) {
                      m_EOF = true;
                  }
              }
              else if (msg.m_msgType == StreamSnapshotMessageType.SCHEMA) {
                  rejoinLog.trace("Got SCHEMA message " + msg.m_blockIndex + " from " +
                          CoreUtils.hsIdToString(msg.m_srcHSId) +
                          " (TargetId " + msg.m_dataTargetId + ")");
                  ByteBuffer block = msg.m_container.b();
                  block.position(StreamSnapshotDataTarget.contentOffset);
                  byte[] schemaBytes = new byte[block.remaining()];
                  block.get(schemaBytes);
                  m_schemas.put(msg.m_tableId, schemaBytes);
              }
              else if (msg.m_msgType == StreamSnapshotMessageType.HASHINATOR) {
                  ByteBuffer block = msg.m_container.b();
                  block.position(StreamSnapshotDataTarget.contentOffset);
                  long version = block.getLong();
                  byte[] hashinatorConfig = new byte[block.remaining()];
                  block.get(hashinatorConfig);
  
                  restoreWork = new HashinatorRestoreWork(version, hashinatorConfig);
              }
              else {
                  // It's normal snapshot data afterwards
                  rejoinLog.trace("Got DATA message " + msg.m_blockIndex + " from " +
                          CoreUtils.hsIdToString(msg.m_srcHSId) +
                          " (TargetId " + msg.m_dataTargetId + ")");
  
                  ByteBuffer block = msg.m_container.b();
  
                  if (!m_schemas.containsKey(msg.m_tableId)) {
                      VoltDB.crashLocalVoltDB("No schema for table with ID " + msg.m_tableId,
                                              false, null);
                  }
  
                  // Get the byte buffer ready to be consumed
                  block.position(StreamSnapshotDataTarget.contentOffset);
                  ByteBuffer nextChunk = getNextChunk(m_schemas.get(msg.m_tableId), block, resultBufferAllocator);
                  m_bytesReceived += nextChunk.remaining();
  
                  restoreWork = new TableRestoreWork(msg.m_tableId, nextChunk);
              }
  
              return restoreWork;
          } finally {
              msg.m_container.discard();
  
              // Queue ack to this block (after the container has been discarded)
              m_ack.ack(msg.m_srcHSId, msg.m_msgType == StreamSnapshotMessageType.END, msg.m_dataTargetId, msg.m_blockIndex);
          }
      }