private static < IN , OUT > SingleOutputStreamOperator < OUT > addOperator ( DataStream < IN > in , AsyncFunction < IN , OUT > func , long timeout , int bufSize , OutputMode mode ) { TypeInformation < OUT > outTypeInfo = TypeExtractor . getUnaryOperatorReturnType ( func , AsyncFunction . class , 0 , 1 , new int [ ] { 1 , 0 } , in . getType ( ) , Utils . getCallLocationName ( ) , true ) ; // create transform AsyncWaitOperator < IN , OUT > operator = new AsyncWaitOperator <> ( in . getExecutionEnvironment ( ) . clean ( func ) , timeout , bufSize , mode ) ; return in . transform ( "async wait operator" , outTypeInfo , operator ) ; }
Add an AsyncWaitOperator .