代码中使用kafka作为数据源的，报错Jobexecutionfailed且指向env.execute()这行，日志中Error:Noentryfoundforconnectionxx

误导&＃xff1a;

如果一直查 Job execution failed 的话&＃xff0c;查到的大部分解决方法都是需要 nc -lk 端口号

其实这个报错的关键是在kafka上&＃xff0c;日志拉下来会看到下面的这一段

日志中的这段&＃xff0c;我遇到的是数字是2147483646&＃xff08;可以参考这篇文章&＃xff09;&＃xff0c;但网上查到的不同数字的解决方式都大同小异

Error: No entry found for connection 2147483646

我的代码情况是&＃xff1a;

Properties properties &＃61; new Properties();
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "consumer-group");
properties.setProperty("key.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("value.deserializer","org.apache.kafka.common.serialization.StringDeserializer");
properties.setProperty("auto.offset.reset", "latest");
DataStream inputStream &＃61; env.addSource(new FlinkKafkaConsumer("hotitems", new SimpleStringSchema(), properties));

解决方式是&＃xff1a;

在kafka的server.properties这个文件中进行修改&＃xff0c;我是修改成下面这样子&＃xff08;ip可以根据自己的需求进行替换&＃xff09;。然后再重启kafka就可以了。

listeners&＃61;PLAINTEXT://localhost:9092
advertised.listeners&＃61;PLAINTEXT://localhost:9092