{"action":"closed","issue":{"url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/179","labels_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/179/labels{/name}","comments_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/179/comments","events_url":"https://api.github.com/repos/SOHU-Co/kafka-node/issues/179/events","html_url":"https://github.com/SOHU-Co/kafka-node/issues/179","id":62722589,"number":179,"title":"High memory usage 1gb+ for 100k+ messages to kafka-node","user":{"login":"mwaarna","id":8648563,"avatar_url":"https://avatars.githubusercontent.com/u/8648563?v=3","gravatar_id":"","url":"https://api.github.com/users/mwaarna","html_url":"https://github.com/mwaarna","followers_url":"https://api.github.com/users/mwaarna/followers","following_url":"https://api.github.com/users/mwaarna/following{/other_user}","gists_url":"https://api.github.com/users/mwaarna/gists{/gist_id}","starred_url":"https://api.github.com/users/mwaarna/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/mwaarna/subscriptions","organizations_url":"https://api.github.com/users/mwaarna/orgs","repos_url":"https://api.github.com/users/mwaarna/repos","events_url":"https://api.github.com/users/mwaarna/events{/privacy}","received_events_url":"https://api.github.com/users/mwaarna/received_events","type":"User","site_admin":false},"labels":[],"state":"closed","locked":false,"assignee":null,"milestone":null,"comments":3,"created_at":"2015-03-18T15:33:43Z","updated_at":"2015-03-19T16:43:45Z","closed_at":"2015-03-19T16:43:45Z","body":"What kind of memory usage is everyone seeing with Kafka node?\r\n\r\nI am sending 4kb messages in bulk(100k+) to kafka node and I am seeing ram usage climb rapidly to 1gb+.\r\n\r\nTesting on Ubuntu 12.04 and Windows 7 64 bit.\r\nBoth running Node V 0.12.0\r\n\r\nKafka version 8.1.1\r\n\r\nSample Node.js code to simulate:\r\n```javascript\r\nvar express = require('express'); \t\t// call express\r\nvar kafka = require('kafka-node');\r\nvar os = require('os');\r\nvar bodyParser = require('body-parser');\r\nvar log4js = require('log4js');\r\n\r\n\r\nvar app = express(); \t\t\t\t// define our app using express\r\nvar iTotalPosts = 0;\r\nvar iTotalPostsSuccess = 0;\r\nvar iTotalPostsError = 0;\r\nvar bProducerReady = false;\r\nvar StartUpTime = new Date();\r\nvar LastSuccessFulConnection = null;\r\nvar dLastErrorEmailSent = null;\r\n\r\nvar hostName = os.hostname();\r\n\r\nconsole.log('process.cwd(): ' + process.cwd());\r\nconsole.log('os.hostname(): ' + hostName);\r\n\r\n\r\nvar config = require('config');\r\n\r\nconsole.log('NODE_CONFIG_DIR: ' + config.util.getEnv('NODE_CONFIG_DIR'));\r\n\r\nlog4js.configure(config.get('logger.configPath'), {});\r\n\r\nvar logger_log4js = log4js.getLogger(config.get('logger.category'));\r\nlogger_log4js.setLevel('DEBUG');\r\nlogger_log4js.debug('Kafka Rest Services Starting!');\r\n\r\n\r\nvar nodemailer = require('nodemailer');\r\n\r\nvar transporter = nodemailer.createTransport({\r\n    host: config.get('mail.options.host'),\r\n    port: config.get('mail.options.port'),\r\n    ignoreTLS: true,\r\n    maxConnections: 5,\r\n    maxMessages: 10\r\n});\r\n\r\n\r\nvar mailOptions = {\r\n    from: config.get('config.fromAddress'), // sender address\r\n    to: config.get('config.toAddress'), // list of receivers\r\n    subject: \"Kafka Post\"\r\n};\r\n\r\n//https://github.com/alexguan/node-zookeeper-client\r\nvar zkOptionsSessionTimeOut = 30000; //Session timeout in milliseconds, defaults to 30 seconds.\r\nvar zkOptionsSpinDelay = 30; //The delay (in milliseconds) between each connection attempts.\r\nvar zkOptionsRetries = 0; //The number of retry attempts for connection loss exception.\r\nvar zkOptions = {\"sessionTimeOut\": zkOptionsSessionTimeOut, \"spinDelay\": zkOptionsSpinDelay, \"retries\": zkOptionsRetries };\r\nvar zkConnectionStr = config.get('config.zkConnectionStr');\r\nvar HighLevelProducer = kafka.HighLevelProducer;\r\nvar client = new kafka.Client(zkConnectionStr, 'NodeJsProducer', zkOptions),\r\n    producer = new HighLevelProducer(client);\r\n\r\nproducer.on('ready', function () {\r\n    console.log(\"Producer to Kafka ready!\");\r\n    var kafkaTopics = ['TestTopic'];\r\n    producer.createTopics(kafkaTopics, false, function (err, data) {\r\n        if (data)\r\n            console.log('kafka topic created Successfully ' + data);\r\n        else if (err)\r\n            console.log('kafka topic Creation Error= ' + err);\r\n        bProducerReady = true;\r\n        LastSuccessFulConnection = new Date();\r\n    });\r\n});\r\n\r\nconsole.log('zkConnectionStr: ' + zkConnectionStr);\r\n\r\napp.use(bodyParser.urlencoded({ extended: true }));\r\napp.use(bodyParser.json());\r\n\r\nvar port = 8888; \t\t// set our port\r\n\r\n// ROUTES FOR OUR API\r\n// =============================================================================\r\nvar router = express.Router(); \t\t\t\t// get an instance of the express Router\r\n\r\nrouter.post('/heartbeat/', function (req, res) {\r\n// get Basic statistics back\r\n    res.json({StartUpTime: StartUpTime,\r\n        iTotalPosts: iTotalPosts,\r\n        iTotalPostsSuccess: iTotalPostsSuccess,\r\n        iTotalPostsError: iTotalPostsError,\r\n        bProducerReady: bProducerReady,\r\n        zkConnectionStr: zkConnectionStr,\r\n        LastSuccessFulConnection: LastSuccessFulConnection});\r\n\r\n});\r\n\r\n//  Kafka Receiver\r\nrouter.post('/kafka/:async', function (req, res) {\r\n\r\n    var bAsync = ( req.params.async.toUpperCase() == \"TRUE\")\r\n    var kafkaTopic = \"TestTopic\";\r\n    iTotalPosts++;\r\n        try {\r\n\r\n            if (bProducerReady) {\r\n                producer.send([{ topic: kafkaTopic, messages: JSON.stringify(req.body)}], function (err, data) {\r\n                    if (data) {\r\n                        LastSuccessFulConnection = new Date();\r\n                        iTotalPostsSuccess++;\r\n                        console.log(data);\r\n\r\n                        if (!bAsync)\r\n                            res.json({KafkaResponse: data});\r\n                    }\r\n                    if (err) {\r\n                        handleError(\"ERROR producer.send \" + (err != null ? err.toString() : \"\"), res, req, bAsync);\r\n                    }\r\n                });\r\n            }\r\n\r\n            producer.on('error', function (err) {\r\n                handleError(\"producer.on 'error' \" + (err != null ? err.toString() : \"\"), res, req, bAsync);\r\n            });\r\n\r\n        }\r\n        catch (e) {\r\n            handleError(\"try catch \" + e.toString(), res, req, bAsync);\r\n        }\r\n\r\n        if (bAsync && bProducerReady)\r\n            res.json({KafkaResponse: \"Async Kafka Request received!\"});\r\n        else if (!bProducerReady)\r\n            handleError(\"Producer Is Not Ready\", res, req, bAsync);\r\n});\r\n\r\nfunction handleError(err, res, req, bAsync) {\r\n    try {\r\n        iTotalPostsError++;\r\n        console.log(err);\r\n        logger_log4js.error(\"ERROR Kafka for \" + JSON.stringify(req.body));\r\n        logger_log4js.error(err);\r\n\r\n        // Only send out 1 error email every 10 minutes\r\n        if (dLastErrorEmailSent == null || ( dLastErrorEmailSent != null && ( new Date().valueOf() - dLastErrorEmailSent.valueOf()) > 600000 )) {\r\n            dLastErrorEmailSent = new Date();\r\n            mailOptions.subject = hostName + \" Error! \" + mailOptions.subject;\r\n            mailOptions.html = JSON.stringify(req.body) + \" Error \" + err;\r\n            transporter.sendMail(mailOptions);// Only Send Error Emails\r\n        }\r\n\r\n        if (!bAsync)\r\n            res.status(500).send(err);\r\n    }\r\n    catch (e) {\r\n        console.log(e);\r\n    }\r\n}\r\n\r\n// REGISTER OUR ROUTES -------------------------------\r\n// all of our routes will be prefixed with /api\r\napp.use('/api', router);\r\napp.use(function (req, res, next) {\r\n    res.status(404).send('That is not the route you are looking for!');\r\n});\r\n\r\n// START THE SERVER\r\n// =============================================================================\r\napp.listen(port, config.get('config.listeningIP'));\r\nconsole.log('Listening on IP ' + config.get('config.listeningIP'));\r\nconsole.log('listening on port:' + port)\r\n```\r\n\r\nSimple Classic ASP to hammer the Node app:\r\n\r\n```javascript\r\nfunction ProduceToKafka(sData, iPostCount, bAsyncRequest )\r\n{\r\n    var bSuccess = true;\r\n    var objHTTP = null;\r\n    var szPostURL = \"http://localhost:8888/api/kafka/\"+bAsyncRequest;\r\n    try\r\n    {\r\n       objHTTP = Server.CreateObject(\"WinHttp.WinHttpRequest.5.1\");\r\n       objHTTP.SetTimeouts(0,      // ResolveTimeout\r\n                            5000,   //ConnectTimeout\r\n                            5000,   //SendTimeout\r\n                            5000);  //ReceiveTimeout\r\n\r\n        for(x=0;x<iPostCount;x++)\r\n        {\r\n            objHTTP.open(\"POST\", szPostURL, false);\r\n            objHTTP.setRequestHeader(\"Content-Type\", \"application/json; charset=utf-8\");\r\n            objHTTP.send(sData);\r\n            \r\n            if ( objHTTP.status != 200 )\r\n            {\r\n                bSuccess = false;\r\n            }\r\n        }\r\n        \r\n    }\r\n    catch(e)\r\n    {\r\n        bSuccess = false;\r\n        Response.Write(\"<br>Error Posting to Kafka = \" + e.description );\r\n    }\r\n    \r\n    if (objHTTP != null)\r\n        objHTTP = null;\r\n    \r\n    return bSuccess;\r\n}\r\n\r\n```"}}