docs: void flattenToSubMessageBuffer(ByteBuffer buf) throws IOException
      {
          // See Serialization Format comment above getSerializedSize().
  
          assert(m_items != null);
          assert(!m_items.isEmpty());
  
          buf.put(VoltDbMessageFactory.FRAGMENT_TASK_ID);
          super.flattenToBuffer(buf);
  
          // Get useful statistics for the header and optional blocks.
          short nInputDepIds = 0;
          short nOutputDepIds = 0;
          short nUnplanned = 0;
          for (FragmentData item : m_items) {
              if (item.m_inputDepId != -1) {
                  // Supporting only one input dep id for now.
                  nInputDepIds++;
              }
              if (item.m_outputDepId != null) {
                  nOutputDepIds++;
              }
              if (item.m_fragmentPlan != null) {
                  nUnplanned++;
              }
          }
  
          // Header block
          buf.putShort((short) m_items.size());
          buf.putShort(nUnplanned);
          buf.put(m_isFinal ? (byte) 1 : (byte) 0);
          buf.put(m_taskType);
          buf.put(m_emptyForRestart ? (byte) 1 : (byte) 0);
          buf.put(m_executedOnPreviousLeader ? (byte) 1 : (byte) 0);
          buf.put(nOutputDepIds > 0 ? (byte) 1 : (byte) 0);
          buf.put(nInputDepIds  > 0 ? (byte) 1 : (byte) 0);
          if (m_procNameToLoad != null) {
              buf.putShort((short) m_procNameToLoad.length);
              buf.put(m_procNameToLoad);
          }
          else {
              buf.putShort((short) -1);
          }
          buf.put(m_perFragmentStatsRecording ? (byte) 1 : (byte) 0);
          buf.put(m_coordinatorTask ? (byte) 1 : (byte) 0);
          // N Partition Transaction PartitionCount
          buf.put(m_nPartTxn ? (byte) 1 : (byte) 0);
          // timestamp for restarted transaction
          buf.putLong(m_restartTimestamp);
  
          buf.putLong(m_lastSpUniqueId);
  
          // Plan Hash block
          for (FragmentData item : m_items) {
              buf.put(item.m_planHash);
          }
  
          for (FragmentData item : m_items) {
              // write statement name
              if (item.m_stmtName == null) {
                  buf.putShort((short) -1);
              }
              else {
                  assert(item.m_stmtName.length <= Short.MAX_VALUE);
                  buf.putShort((short) item.m_stmtName.length);
                  buf.put(item.m_stmtName);
              }
          }
  
          // Parameter set block
          for (FragmentData item : m_items) {
              buf.putInt(item.m_parameterSet.remaining());
              buf.put(item.m_parameterSet.asReadOnlyBuffer());
          }
  
          // Optional output dependency ID block
          if (nOutputDepIds > 0) {
              for (FragmentData item : m_items) {
                  buf.putInt(item.m_outputDepId);
              }
          }
  
          // Optional input dependency ID block
          if (nInputDepIds > 0) {
              for (FragmentData item : m_items) {
                  if (item.m_inputDepId == -1) {
                      buf.putShort((short) 0);
                  } else {
                      buf.putShort((short) 1);
                      buf.putInt(item.m_inputDepId);
                  }
              }
          }
  
          // write procedure name
          if (m_procedureName == null) {
              buf.putShort((short) -1);
          }
          else {
              assert(m_procedureName.length <= Short.MAX_VALUE);
              buf.putShort((short) m_procedureName.length);
              buf.put(m_procedureName);
          }
  
          // ints for batch context
          buf.putInt(m_currentBatchIndex);
  
          // put byte flag for timeout value and 4 bytes integer value if specified
          if (m_batchTimeout == BatchTimeoutOverrideType.NO_TIMEOUT) {
              buf.put(BatchTimeoutOverrideType.NO_OVERRIDE_FOR_BATCH_TIMEOUT.getValue());
          } else {
              buf.put(BatchTimeoutOverrideType.HAS_OVERRIDE_FOR_BATCH_TIMEOUT.getValue());
              buf.putInt(m_batchTimeout);
          }
  
          buf.putShort((short) m_involvedPartitions.size());
          for (int pid : m_involvedPartitions) {
              buf.putInt(pid);
          }
  
          if (m_initiateTaskBuffer != null) {
              ByteBuffer dup = m_initiateTaskBuffer.duplicate();
              buf.putInt(dup.remaining());
              buf.put(dup);
          } else {
              buf.putInt(0);
          }
  
          // Unplanned item block
          for (short index = 0; index < m_items.size(); index++) {
              // Each unplanned item gets:
              //  - an index (2)
              //  - a size (4)
              //  - buffer for the fragment plan string
              //  - a size(4)
              //  - buffer for the statement text
              FragmentData item = m_items.get(index);
              if (item.m_fragmentPlan != null) {
                  buf.putShort(index);
                  buf.putInt(item.m_fragmentPlan.length);
                  buf.put(item.m_fragmentPlan);
  
                  assert(item.m_stmtText != null);
                  buf.putInt(item.m_stmtText.length);
                  buf.put(item.m_stmtText);
              }
          }
  
      }