public boolean callProcedure ( InternalConnectionContext caller , Function < Integer , Boolean > backPressurePredicate , InternalConnectionStatsCollector statsCollector , ProcedureCallback procCallback , String proc , Object ... fieldList ) { Procedure catProc = InvocationDispatcher . getProcedureFromName ( proc , getCatalogContext ( ) ) ; if ( catProc == null ) { String fmt = "Cannot invoke procedure %s from streaming interface %s. Procedure not found." ; m_logger . rateLimitedLog ( SUPPRESS_INTERVAL , Level . ERROR , null , fmt , proc , caller ) ; m_failedCount . incrementAndGet ( ) ; return false ; } StoredProcedureInvocation task = new StoredProcedureInvocation ( ) ; task . setProcName ( proc ) ; task . setParams ( fieldList ) ; try { task = MiscUtils . roundTripForCL ( task ) ; } catch ( Exception e ) { String fmt = "Cannot invoke procedure %s from streaming interface %s. failed to create task." ; m_logger . rateLimitedLog ( SUPPRESS_INTERVAL , Level . ERROR , null , fmt , proc , caller ) ; m_failedCount . incrementAndGet ( ) ; return false ; } int [ ] partitions = null ; try { partitions = InvocationDispatcher . getPartitionsForProcedure ( catProc , task ) ; } catch ( Exception e ) { String fmt = "Can not invoke procedure %s from streaming interface %s. Partition not found." ; m_logger . rateLimitedLog ( SUPPRESS_INTERVAL , Level . ERROR , e , fmt , proc , caller ) ; m_failedCount . incrementAndGet ( ) ; return false ; } boolean mp = ( partitions [ 0 ] == MpInitiator . MP_INIT_PID ) || ( partitions . length > 1 ) ; final InternalClientResponseAdapter adapter = mp ? m_adapters . get ( MpInitiator . MP_INIT_PID ) : m_adapters . get ( partitions [ 0 ] ) ; InternalAdapterTaskAttributes kattrs = new InternalAdapterTaskAttributes ( caller , adapter . connectionId ( ) ) ; final AuthUser user = getCatalogContext ( ) . authSystem . getImporterUser ( ) ; if ( ! adapter . createTransaction ( kattrs , proc , catProc , procCallback , statsCollector , task , user , partitions , false , backPressurePredicate ) ) { m_failedCount . incrementAndGet ( ) ; return false ; } m_submitSuccessCount . incrementAndGet ( ) ; return true ; }
Use null backPressurePredicate for no back pressure