我希望处理来自Kafka的消息，然后提交该消息，Flink使用并处理所有消息后，结束作业，使用任务管理器和心跳升级进程

StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();
    KafkaSource<String> source = KafkaSource.<String>builder()
            .setBootstrapServers(address)
            .setTopics(inputTopic)
            .setGroupId(consumerGroup)
            .setStartingOffsets(OffsetsInitializer.earliest())
            .setValueOnlyDeserializer(new SimpleStringSchema())
            .setProperty("enable.auto.commit", "true")
            .setProperty("commit.offsets.on.checkpoint", "true")
            .build();

    DataStream<String> stream = environment.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
    ObjectMapper mapper = new ObjectMapper();
    stream.map((value) -> {