KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("localhost:9092")
    .setTopics("source")
    .setGroupId("my-group")
    .setStartingOffsets(OffsetsInitializer.latest())
    .setValueOnlyDeserializer(new SimpleStringSchema())
    .build();

DataStream<String> sourceStream = env.fromSource(
        source, WatermarkStrategy.forMonotonousTimestamps(), "Kafka Source")
    .uid("kafkasourceuid");