public class WordCount {

    // Job definition
    public static void main(String[] args) throws Exception {
        // Some initialization code...

        // Source
        DataStream<String> text = env.fromSource(builder.build(), WatermarkStrategy.noWatermarks(), "file-input");

        // Aggregate
        DataStream<Tuple2<String, Integer>> counts =
                text.flatMap(new Tokenizer())
                        .name("tokenizer")
                        .keyBy(value -> value.f0)
                        // User codes
                        .flatMap(new Counter())
                        .name("counter");

        // Sink
        counts.print().name("print-sink");
        
        env.execute("WordCount");
    }

    // Core processing user function
    public static final class Counter extends RichFlatMapFunction<Tuple2<String, Integer>, Tuple2<String, Integer>> {

        // Initialization omitted.
        private transient ValueState<Integer> wordCounter;
        
        // Synchronous state API usage
        @Override
        public void flatMap(Tuple2<String, Integer> value, Collector<Tuple2<String, Integer>> out) throws Exception {
            Integer val = wordCounter.value();
            int updated = (val == null ? 1 : val + 1);
            wordCounter.update(updated);
            out.collect(Tuple2.of(value.f0, updated)); 
        }

        // Asynchronous state API usage
        @Override
        public void flatMap(Tuple2<String, Integer> value, Collector<Tuple2<String, Integer>> out) throws Exception {
            AtomicInteger updated = new AtomicInteger(0);
            wordCounter.value().thenthenCompose(
                    val -> {
                        if (val == null) {
                            updated.set(1);
                        } else {
                            updated.set(val + 1);
                        }
                        return wordCounter.update(updated.get());
                    }
            ).thenthenAccept(
                    empty -> {
                        out.collect(Tuple2.of(value.f0, updated.get()));
                    }
            );
        }
        
    }
}