brewery_example.java

tableEnv.connect(new Kafka()
        .version("0.11")
        .topic(params.getRequired("read-topic"))
        .property("bootstrap.servers", params.getRequired("bootstrap.servers")))
        .withSchema(new Schema()
                .field("sensor", Types.STRING())
                .field("temp", Types.LONG())
                .field("ts", Types.SQL_TIMESTAMP())
                .rowtime(new Rowtime()
                        .timestampsFromSource()
                        .watermarksPeriodicBounded(1000)
                )
        )
        .withFormat(new Json().deriveSchema())
        .inAppendMode()
        .registerTableSource("sourceTopic");