Hello team, I am running a simple code of Flink Kafka Source, but it does not continue fetch data from the Kafka topic after fetching few records. And no error message printed in the TaskManager. Could you help on this? Any thoughts? Flink runtime : 1.6.0 flink-connector-kafka: 1.6.0 Kafka-client : 3.1.1, in order to match the version of Kafka Broker. final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); final String kafkaBrokers = "xxxxx:xx"; final String topicName = "xxxx"; final String consumerGroup = "Flink-justprint-group-12345"; KafkaSource<String> source = KafkaSource.<String>builder() .setBootstrapServers(kafkaBrokers) .setTopics(topicName) .setGroupId(consumerGroup) .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST)) .setValueOnlyDeserializer(new SimpleStringSchema()) .build(); env.getCheckpointConfig().disableCheckpointing(); DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source"); stream.map(item -> item.length()%5) .map(item -> DateTime.now().toString() + ":" + item) .print(); stream.executeAndCollect();

stream.executeAndCollect();

jumps out, have you tried

env.execute();

instead?

let me try it now.

👍

stream.executeAndCollect()

is something that you generally use in a test.