# SparkStreaming消费Kafka消息失败，报错"Couldn't find leader offsets"
#### 问题背景与现象
使用SparkStreaming来消费Kafka中指定Topic的消息时，发现无法从Kafka中获取到数据。
提示如下错误： Couldn't find leader offsets。
![](https://support.huaweicloud.com/trouble-mrs/zh-cn_image_0264281647.png "点击放大")
#### 可能原因
- Kafka服务异常。
- 网络异常。
- Kafka Topic异常。
 
#### 原因分析
1. 通过Manager页面，查看Kafka集群当前状态，发现状态为"良好"，且监控指标内容显示正确。
2. 查看SparkStreaming日志中提示错误的Topic信息。 
   执行Kafka相关命令，获取Topic分布信息和副本同步信息，观察返回结果。
   **kafka-topics.sh --describe --zookeeper** *\<zk_host:port/chroot\>* **--topic** *\<topic name\>*
   如下所示，发现对应Topic状态正常。所有Partition均存在正常Leader信息。
   图1Topic分布信息和副本同步信息   
   ![](https://support.huaweicloud.com/trouble-mrs/zh-cn_image_0264281912.png "点击放大")
   
   
3. 检查客户端与Kafka集群网络是否连通，若网络不通协调网络组进行处理。
4. 通过SSH登录Kafka Broker。 
   通过**cd /var/log/Bigdata/kafka/broker**命令进入日志目录。
   查看server.log发现如下日志抛出java.lang.OutOfMemoryError: Direct buffer memory。
   ```
   2018-05-30 12:02:00,246 | ERROR | [kafka-network-thread-6-PLAINTEXT-3] | Processor got uncaught exception. | kafka.network.Processor (Logging.scala:103) 
    java.lang.OutOfMemoryError: Direct buffer memory
    at java.nio.Bits.reserveMemory(Bits.java:694)
    at java.nio.DirectByteBuffer.<init>(DirectByteBuffer.java:123)
    at java.nio.ByteBuffer.allocateDirect(ByteBuffer.java:311)
    at sun.nio.ch.Util.getTemporaryDirectBuffer(Util.java:241)
    at sun.nio.ch.IOUtil.read(IOUtil.java:195)
    at sun.nio.ch.SocketChannelImpl.read(SocketChannelImpl.java:380)
   ```
   at org.apache.kafka.common.network.PlaintextTransportLayer.read(PlaintextTransportLayer.java:110)
   
   
5. 通过Manager页面，查看当前Kafka集群配置。 
   - MRS Manager界面操作：登录MRS Manager，选择"服务管理 \> Kafka \> 服务配置"，"参数类别"设置为"全部配置"，发现"KAFKA_JVM_PERFORMANCE_OPTS"的中"-XX:MaxDirectMemorySize"值为"1G"。
   
   - FusionInsight Manager界面操作：登录FusionInsight Manager。选择"集群 \> 服务 \> Kafka"，单击"配置"，选择"全部配置"，发现"KAFKA_JVM_PERFORMANCE_OPTS"的中"-XX:MaxDirectMemorySize"值为"1G"。
   
   
   
   
6. 直接内存配置过小导致报错，而且一旦直接内存溢出，该节点将无法处理新请求，会导致其他节点或者客户端访问超时失败。
 
#### 解决办法
1. 登录到Manager，进入Kafka配置页面。
2. 选择"全部配置"，搜索并修改KAFKA_JVM_PERFORMANCE_OPTS的值。
3. 保存配置，并重启配置过期的服务或者实例。