public FragmentResponseMessage processFragmentTask ( SiteProcedureConnection siteConnection ) { final FragmentResponseMessage currentFragResponse = new FragmentResponseMessage ( m_fragmentMsg , m_initiator . getHSId ( ) ) ; currentFragResponse . setStatus ( FragmentResponseMessage . SUCCESS , null ) ; for ( int frag = 0 ; frag < m_fragmentMsg . getFragmentCount ( ) ; frag ++ ) { final long fragmentId = VoltSystemProcedure . hashToFragId ( m_fragmentMsg . getPlanHash ( frag ) ) ; // equivalent to dep.depId: // final int outputDepId = m_fragmentMsg.getOutputDepId(frag); final VoltTrace . TraceEventBatch traceLog = VoltTrace . log ( VoltTrace . Category . SPSITE ) ; if ( traceLog != null ) { traceLog . add ( ( ) -> VoltTrace . beginDuration ( "runfragmenttask" , "txnId" , TxnEgo . txnIdToString ( getTxnId ( ) ) , "partition" , Integer . toString ( siteConnection . getCorrespondingPartitionId ( ) ) , "fragmentId" , String . valueOf ( fragmentId ) ) ) ; } ParameterSet params = m_fragmentMsg . getParameterSetForFragment ( frag ) ; try { // run the overloaded sysproc planfragment. pass an empty dependency // set since remote (non-aggregator) fragments don't receive dependencies. final DependencyPair dep = siteConnection . executeSysProcPlanFragment ( m_txnState , m_inputDeps , fragmentId , params ) ; // @Shutdown returns null, handle it here if ( dep != null ) { currentFragResponse . addDependency ( dep ) ; } } catch ( final EEException | SQLException | ReplicatedTableException e ) { hostLog . l7dlog ( Level . TRACE , LogKeys . host_ExecutionSite_ExceptionExecutingPF . name ( ) , new Object [ ] { Encoder . hexEncode ( m_fragmentMsg . getFragmentPlan ( frag ) ) } , e ) ; currentFragResponse . setStatus ( FragmentResponseMessage . UNEXPECTED_ERROR , e ) ; addDependencyToFragment ( currentFragResponse ) ; break ; } catch ( final SerializableException e ) { // Note that with SerializableException, the error code here might get changed before // the client/user sees it. It really just needs to indicate failure. // // Key point here vs the next catch block for VAE is to not wrap the subclass of // SerializableException here to preserve it during the serialization. // currentFragResponse . setStatus ( FragmentResponseMessage . USER_ERROR , e ) ; addDependencyToFragment ( currentFragResponse ) ; break ; } catch ( final VoltAbortException e ) { currentFragResponse . setStatus ( FragmentResponseMessage . USER_ERROR , new SerializableException ( CoreUtils . throwableToString ( e ) ) ) ; addDependencyToFragment ( currentFragResponse ) ; break ; } if ( traceLog != null ) { traceLog . add ( VoltTrace :: endDuration ) ; } } // we should never rollback DR buffer for MP sysprocs because we don't report the DR buffer size and therefore don't know if it is empty or not. currentFragResponse . setDrBufferSize ( 1 ) ; return currentFragResponse ; }
modified to work in the new world