Flink 接收不到Kafka数据

Question

Kafka集群在虚拟机上，能正常通信能互相ping本地consumer消费者程序正常消费，flink程序不能消费不能打印数据，也没有错误提示

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


public class Cityweather {
    public static void main(String[] args) throws Exception {
        // 创建流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置checkpoint
        env.enableCheckpointing(1000, CheckpointingMode.AT_LEAST_ONCE);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000);

        // 创建 KafkaSource
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
                .setBootstrapServers("hadoop100:9092")
                .setTopics("POC_test")
                .setGroupId("my-groupsdads")
                .setStartingOffsets(OffsetsInitializer.earliest())
                .setValueOnlyDeserializer(new SimpleStringSchema())
                .build();

        // 从 Kafka 读取数据并创建 DataStream
        DataStreamSource<String> weathertest = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "weather_test");

//        System.out.println(weatherTest);
        weathertest.map(value -> {
            System.out.println(value);
            Thread.sleep(10000);
            return value;
        });

        // 打印数据流
//        weathertest.print();

        // 执行任务
        env.execute("CityWeather");
    }
}

Answer 1

• 你可以参考下这个问题的回答, 看看是否对你有帮助, 链接: https://ask.csdn.net/questions/7491501
• 我还给你找了一篇非常好的博客，你可以看看是否有帮助，链接：flink连上了Kafka 却接收不到数据
• 除此之外, 这篇博客: Flink学习笔记：Kafka集群搭建中的 环境准备、 部分也许能够解决你的问题, 你可以仔细阅读以下内容或跳转源博客中阅读:
  
• • 安装好JDK并设置好环境变量
  • 关闭防火墙
  • ssh免密登录
  • 搭建zookeeper集群
    [集群搭建方式见]Zookeeper集群搭建
• 您还可以看一下 成宝老师的2020最新 Flink入门及进阶课程课程中的 Flink 常用API详解之KafkaSourceTwo小节, 巩固相关知识点