{"action":"created","issue":{"url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/1107","repository_url":"https://api.github.com/repos/SOHU-Co/kafka-node","labels_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/1107/labels{/name}","comments_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/1107/comments","events_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/1107/events","html_url":"https://github.com/SOHU-Co/kafka-node/issues/1107","id":372998913,"node_id":"MDU6SXNzdWUzNzI5OTg5MTM=","number":1107,"title":"kafka client is sending request to broker which went down","user":{"login":"shashanksah","id":3831391,"node_id":"MDQ6VXNlcjM4MzEzOTE=","avatar_url":"https://avatars0.githubusercontent.com/u/3831391?v=4","gravatar_id":"","url":"https://api.github.com/users/shashanksah","html_url":"https://github.com/shashanksah","followers_url":"https://api.github.com/users/shashanksah/followers","following_url":"https://api.github.com/users/shashanksah/following{/other_user}","gists_url":"https://api.github.com/users/shashanksah/gists{/gist_id}","starred_url":"https://api.github.com/users/shashanksah/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/shashanksah/subscriptions","organizations_url":"https://api.github.com/users/shashanksah/orgs","repos_url":"https://api.github.com/users/shashanksah/repos","events_url":"https://api.github.com/users/shashanksah/events{/privacy}","received_events_url":"https://api.github.com/users/shashanksah/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":1,"created_at":"2018-10-23T13:25:52Z","updated_at":"2018-10-23T14:45:04Z","closed_at":null,"author_association":"NONE","body":"## Bug Report\r\nWhen a broker host goes down/restarts, the kafka client keeps sending the new requests to the same hosts. As a result, the requests are failing with this error: Request timed out after 30000ms\r\n### Environment\r\n\r\n* Node version: v6.8.1\r\n* Kafka-node version: 3.0.1\r\n* Kafka version: 2.11-2.0.0\r\n\r\n### For specific cases also provide\r\n* Number of Brokers: 3\r\n* Number partitions for topic: 10\r\n\r\n### Include Sample Code to reproduce behavior\r\n\r\n```js\r\n\r\n\"clusterConfig\" : {\r\n    \"kafkaHost\": \"localhost:9092,localhost:9093,localhost:9094\",\r\n    \"autoConnect\": true\r\n  }\r\n...\r\nlet kafkaClient = new kafka.KafkaClient(clusterConfig);\r\n      producer = new kafka.HighLevelProducer(kafkaClient, cluster.producerConfig);\r\n      Promise.promisifyAll(producer);\r\n...\r\nproducer.sendAsync([eventPayload])\r\n            .then(function (data) {\r\n              let topicName = eventPayload.topic;\r\n              let payLoadSize = (eventPayload || '').length;\r\n              logger.eventInfo(topicName, payLoadSize, source);\r\n            })\r\n            .catch(function (e) {\r\n              logger.produceFailedEvent(eventPayload, source);\r\n              throw Error.getErrorObject(errorType, e, topic, source);\r\n            });\r\n```\r\n\r\n### Include output with [Debug](https://github.com/SOHU-Co/kafka-node#how-do-i-debug-an-issue) turned on\r\n\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient compressing messages if needed +787ms\r\n  kafka-node:KafkaClient missing apiSupport waiting until broker is ready... +1ms\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +217ms\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9093 +1s\r\n  kafka-node:KafkaClient compressing messages if needed +528ms\r\n  kafka-node:KafkaClient missing apiSupport waiting until broker is ready... +1ms\r\n\r\n\r\nThanks for your contribution!"},"comment":{"url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/comments/432276967","html_url":"https://github.com/SOHU-Co/kafka-node/issues/1107#issuecomment-432276967","issue_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/1107","id":432276967,"node_id":"MDEyOklzc3VlQ29tbWVudDQzMjI3Njk2Nw==","user":{"login":"shashanksah","id":3831391,"node_id":"MDQ6VXNlcjM4MzEzOTE=","avatar_url":"https://avatars0.githubusercontent.com/u/3831391?v=4","gravatar_id":"","url":"https://api.github.com/users/shashanksah","html_url":"https://github.com/shashanksah","followers_url":"https://api.github.com/users/shashanksah/followers","following_url":"https://api.github.com/users/shashanksah/following{/other_user}","gists_url":"https://api.github.com/users/shashanksah/gists{/gist_id}","starred_url":"https://api.github.com/users/shashanksah/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/shashanksah/subscriptions","organizations_url":"https://api.github.com/users/shashanksah/orgs","repos_url":"https://api.github.com/users/shashanksah/repos","events_url":"https://api.github.com/users/shashanksah/events{/privacy}","received_events_url":"https://api.github.com/users/shashanksah/received_events","type":"User","site_admin":false},"created_at":"2018-10-23T14:45:04Z","updated_at":"2018-10-23T14:45:04Z","author_association":"NONE","body":" Given below are the logs with the corresponding hosts they are going to connect. The first calls was successful, second failed and third was successful. In the below case localhost:9092 was down.\r\n\r\nkafka-node:KafkaClient kafka-node-client reconnecting to localhost:9092 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9092 +1s\r\n  kafka-node:KafkaClient compressing messages if needed +76ms\r\nshashanksah actual broker!!!!![BrokerWrapper localhost:9094 (connected: true) (ready: true) (idle: false) (needAuthentication: false) (authenticated: false)]\r\nerror:  index_name=undefined, uuid=706965d0-d6cf-11e8-85ce-5f32fbd56a5d, deployment_tag=NA, instance_tag=NA, machine_ip=NA, log_type=openapi_rpc_service, log_genre=error, err_type=event_push_error, err_message=<html>\r\n<head><title>503 Service Temporarily Unavailable</title></head>\r\n<body bgcolor=\"white\">\r\n<center><h1>503 Service Temporarily Unavailable</h1></center>\r\n</body>\r\n</html>\r\n, event=review_insert\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9092 +928ms\r\n  kafka-node:KafkaClient compressing messages if needed +422ms\r\nshashanksah actual broker!!!!![BrokerWrapper localhost:9092 (connected: false) (ready: false) (idle: false) (needAuthentication: false) (authenticated: false)]\r\n  kafka-node:KafkaClient missing apiSupport waiting until broker is ready... +1ms\r\nerror:  index_name=undefined, uuid=711d7390-d6cf-11e8-85ce-5f32fbd56a5d, deployment_tag=NA, instance_tag=NA, machine_ip=NA, log_type=openapi_rpc_service, log_genre=error, err_type=event_push_error, err_message=<html>\r\n<head><title>503 Service Temporarily Unavailable</title></head>\r\n<body bgcolor=\"white\">\r\n<center><h1>503 Service Temporarily Unavailable</h1></center>\r\n</body>\r\n</html>\r\n, event=review_insert\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9092 +583ms\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9092 +1s\r\n  kafka-node:KafkaClient compressing messages if needed +280ms\r\nshashanksah actual broker!!!!![BrokerWrapper localhost:9093 (connected: true) (ready: true) (idle: false) (needAuthentication: false) (authenticated: false)]\r\nerror:  index_name=undefined, uuid=7239e010-d6cf-11e8-85ce-5f32fbd56a5d, deployment_tag=NA, instance_tag=NA, machine_ip=NA, log_type=openapi_rpc_service, log_genre=error, err_type=event_push_error, err_message=<html>\r\n<head><title>503 Service Temporarily Unavailable</title></head>\r\n<body bgcolor=\"white\">\r\n<center><h1>503 Service Temporarily Unavailable</h1></center>\r\n</body>\r\n</html>\r\n, event=review_insert\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9092 +723ms\r\nerror:  index_name=undefined, uuid=72cdbba0-d6cf-11e8-85ce-5f32fbd56a5d, deployment_tag=NA, instance_tag=NA, machine_ip=NA, log_type=openapi_rpc_service, log_genre=error, err_type=event_produce_error, err_type=ProducerError, message=Request timed out after 3000ms, event=review_insert, source=review_service, event=review_insert\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9092 +1s\r\n  kafka-node:KafkaClient kafka-node-client reconnecting to localhost:9092 +1s\r\n\r\nThe issue is that it get the ip of host which is down and then waits on that host to be ready (for 30secs) and then fails the sent request. If we change the function ensureBrokerReady in the kafkaClient.js file in kafka-node to something like given below, then this issue does not occurs:\r\n\r\n```\r\n  const ensureBrokerReady = async.ensureAsync((leader, callback) => {\r\n    let broker = this.brokerForLeader(leader, longpolling);\r\n    console.log(\"shashanksah actual broker!!!!!\" + broker);\r\n    if (!broker.isReady()) {\r\n      this.refreshBrokerMetadata();\r\n      broker = this.brokerForLeader(leader, longpolling);\r\n      //console.log(\"shashanksah broker not ready so refresh and retry!!!!!\");\r\n    }\r\n    if (!broker.isReady()) {\r\n      //console.log(\"shashanksah !broker.isReady\");\r\n      logger.debug('missing apiSupport waiting until broker is ready...');\r\n      this.waitUntilReady(broker, callback);\r\n    } else {\r\n      callback(null);\r\n    }\r\n  }); \r\n```\r\nPlease check and tell if the RCA is correct.\r\n\r\nThanks,\r\nShashank"}}
{"actor":{"display_login":"shashanksah"}}