code: <OUT, K> SingleOutputStreamOperator<OUT> build(
  			final TypeInformation<OUT> outTypeInfo,
  			final PatternProcessFunction<IN, OUT> processFunction) {
  
  		checkNotNull(outTypeInfo);
  		checkNotNull(processFunction);
  
  		final TypeSerializer<IN> inputSerializer = inputStream.getType().createSerializer(inputStream.getExecutionConfig());
  		final boolean isProcessingTime = inputStream.getExecutionEnvironment().getStreamTimeCharacteristic() == TimeCharacteristic.ProcessingTime;
  
  		final boolean timeoutHandling = processFunction instanceof TimedOutPartialMatchHandler;
  		final NFACompiler.NFAFactory<IN> nfaFactory = NFACompiler.compileFactory(pattern, timeoutHandling);
  
  		final CepOperator<IN, K, OUT> operator = new CepOperator<>(
  			inputSerializer,
  			isProcessingTime,
  			nfaFactory,
  			comparator,
  			pattern.getAfterMatchSkipStrategy(),
  			processFunction,
  			lateDataOutputTag);
  
  		final SingleOutputStreamOperator<OUT> patternStream;
  		if (inputStream instanceof KeyedStream) {
  			KeyedStream<IN, K> keyedStream = (KeyedStream<IN, K>) inputStream;
  
  			patternStream = keyedStream.transform(
  				"CepOperator",
  				outTypeInfo,
  				operator);
  		} else {
  			KeySelector<IN, Byte> keySelector = new NullByteKeySelector<>();
  
  			patternStream = inputStream.keyBy(keySelector).transform(
  				"GlobalCepOperator",
  				outTypeInfo,
  				operator
  			).forceNonParallel();
  		}
  
  		return patternStream;
  	}
docs: Creates a data stream containing results of {@link PatternProcessFunction} to fully matching event patterns.
  
  @param processFunction function to be applied to matching event sequences
  @param outTypeInfo output TypeInformation of
  {@link PatternProcessFunction#processMatch(Map, PatternProcessFunction.Context, Collector)}
  @param <OUT> type of output events
  @return Data stream containing fully matched event sequence with applied {@link PatternProcessFunction}