messageStream.map(new MapFunction<String, String>() {

                        @Override
                        public String map(String value) throws Exception {
                                JCas jCas = JCasFactory.createJCas();
                                jCas.setDocumentText(value);
                                jCas.setDocumentLanguage("en");

                    AnalysisEngine seg = AnalysisEngineFactory.createEngine(
                                    StanfordSegmenter.class);

                    AnalysisEngine ner = AnalysisEngineFactory.createEngine(
                                    StanfordNamedEntityRecognizer.class);

                    AnalysisEngine parser = AnalysisEngineFactory.createEngine(
                                    StanfordParser.class);

                                seg.process(jCas);
                                ner.process(jCas);
                                parser.process(jCas);

                                return "message length received: " + jCas.getDocumentText().length();
                        }
                }).rebalance().print();