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 ) ; } }
Process a message pulled off from the network thread and discard the container once it s processed .