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;
    }
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