1.概述
近日flinksql作业出现不稳定，查看kafka侧以及flink日志，找到了出现波动时间段附近的日志中有这个信息：
org.apache.flink.kafka.shaded.org.apache.kafka.clients.FetchSessionHandler [] - [ConsumerclientId=consumer-kafka_alarm_10013-2, groupId=kafka_alarm_10013] Errorsendingfetchrequest (sessionId=836884100, epoch=4625) tonode103: {}. org.apache.flink.kafka.shaded.org.apache.kafka.common.errors.DisconnectException: null
初一看应该链接问题，检查kafka集群发现生产流量正常，消费流量出现降，这应该就是flink消费端链接kafka出现了问题。
2.解决
在stackoverflow上发现了同样错误的问题
setting logging.level.org.apache.kafka.*=DEBUG logs to DEBUG shows:
DEBUGorg.apache.kafka.clients.NetworkClient-Disconnectingfromnode1duetorequesttimeout. DEBUGorg.apache.kafka.clients.consumer.internals.ConsumerNetworkClient-CancelledrequestwithheaderRequestHeader(apiKey=FETCH, apiVersion=11, clientId=consumer-1, correlationId=183) duetonode1beingdisconnectedDEBUGorg.apache.kafka.clients.NetworkClient-GiveupsendingmetadatarequestsincenonodeisavailableINFOorg.apache.kafka.clients.FetchSessionHandler-Errorsendingfetchrequest (sessionId=INVALID, epoch=INITIAL) tonode1: {}. org.apache.kafka.common.errors.DisconnectException: nullDEBUGorg.apache.kafka.clients.NetworkClient-Giveupsendingmetadatarequestsincenonodeisavailable
通过分析debug日志给出了如下答复：增加request.timeout.ms参数(default 30000)
props.put(ConsumerConfig.REQUEST_TIMEOUT_MS_CONFIG, "60000");
增加这个参数应该可以解决链接问题，但为何产生这个异常呢。
3.源码分析
根据问题日志找到该类org.apache.kafka.clients.FetchSessionHandler
在这个类里搜索出错信息，发现了这个方法handleError
/*** Handle an error sending the prepared request.*处理发送已准备的请求的错误* When a network error occurs, we close any existing fetch session on our next request,* and try to create a new session.*当发生网络故障时，在下一个请求中关闭任何已存在的fetch会话，并尝试创建一个新会话* @param t The exception.*/publicvoidhandleError(Throwablet) { log.info("Error sending fetch request {} to node {}: {}.", nextMetadata, node, t.toString()); nextMetadata=nextMetadata.nextCloseExisting(); }
说明当网络故障时，就会调用改方法，我们来看是谁调用了该方法
找到该类org.apache.kafka.clients.consumer.internals.Fetcher
/*** Set-up a fetch request for any node that we have assigned partitions for which doesn't already have* an in-flight fetch or pending fetch data.*为我们已为其分配分区且尚未有正在进行中的提取或待处理的提取数据的任何节点设置提取请求* @return number of fetches sent*/publicsynchronizedintsendFetches() { Map<Node, FetchSessionHandler.FetchRequestData>fetchRequestMap=prepareFetchRequests(); for (Map.Entry<Node, FetchSessionHandler.FetchRequestData>entry : fetchRequestMap.entrySet()) { finalNodefetchTarget=entry.getKey(); finalFetchSessionHandler.FetchRequestDatadata=entry.get创建Value(); // todo:创建获取数据的请求finalFetchRequest.Builderrequest=FetchRequest.Builder .forConsumer(this.maxWaitMs, this.minBytes, data.toSend()) .isolationLevel(isolationLevel) .setMaxBytes(this.maxBytes) .metadata(data.metadata()) .toForget(data.toForget()); if (log.isDebugEnabled()) { log.debug("Sending {} {} to broker {}", isolationLevel, data.toString(), fetchTarget); } // todo:发送所有requestclient.send(fetchTarget, request) // todo:发送完成后调用这个回调函数 .addListener(newRequestFutureListener<ClientResponse>() { publicvoidonSuccess(ClientResponseresp) { synchronized (Fetcher.this) { ("unchecked") FetchResponse<Records>response= (FetchResponse<Records>) resp.responseBody(); FetchSessionHandlerhandler=sessionHandler(fetchTarget.id()); if (handler==null) { log.error("Unable to find FetchSessionHandler for node {}. Ignoring fetch response.", fetchTarget.id()); return; } if (!handler.handleResponse(response)) { return; } Set<TopicPartition>partitions=newHashSet<>(response.responseData().keySet()); FetchResponseMetricAggregatormetricAggregator=newFetchResponseMetricAggregator(sensors, partitions); for (Map.Entry<TopicPartition, FetchResponse.PartitionData<Records>>entry : response.responseData().entrySet()) { TopicPartitionpartition=entry.getKey(); longfetchOffset=data.sessionPartitions().get(partition).fetchOffset; FetchResponse.PartitionData<Records>fetchData=entry.getValue(); log.debug("Fetch {} at offset {} for partition {} returned fetch data {}", isolationLevel, fetchOffset, partition, fetchData); completedFetches.add(newCompletedFetch(partition, fetchOffset, fetchData, metricAggregator, resp.requestHeader().apiVersion())); } sensors.fetchLatency.record(resp.requestLatencyMs()); } } // todo:调用异常时走这个函数，这个函数就调用了handleError方法publicvoidonFailure(RuntimeExceptione) { synchronized (Fetcher.this) { FetchSessionHandlerhandler=sessionHandler(fetchTarget.id()); if (handler!=null) { handler.handleError(e); } } } }); }
从源码中发现，在创建完FetchRequest之后，如果出现异常就会回调onFailure，这个函数就调用了handleError方法。
所以这个异常就是在获取请求时发生了网络抖动，造成日志中的错误。
如有分析错误，欢迎指正
拜了个拜
