Error while running consumer about amazon-kinesis-client-nodejs

[niketh90]

Then it will be a good idea to enable debug logs and send logs to aws support for analysis.

[MariaTolstov]

@sapinder
Did you find any solutions?

[ps185130]

We are seeing similar issue while running consumer application using NodeJS aws-kcl library with version 0.7.0 , the simple consumer app fails after some time.

---

Error Log :
Oct 22, 2018 7:47:35 PM com.amazonaws.services.kinesis.multilang.LineReaderTask call
INFO: Stopping: Reading next message from STDIN for shardId-000000000001
Oct 22, 2018 7:47:35 PM com.amazonaws.services.kinesis.multilang.DrainChildSTDERRTask handleLine
SEVERE: Received error line from subprocess [Error: Kinesis Client Library is in the invalid state. Cannot proceed further.] for shard shardId-000000000001
Error: Kinesis Client Library is in the invalid state. Cannot proceed further.
Oct 22, 2018 7:47:35 PM com.amazonaws.services.kinesis.multilang.DrainChildSTDERRTask handleLine
SEVERE: Received error line from subprocess [ at KCLManager._handleStateInput (/usr/src/app/node_modules/aws-kcl/lib/kcl/kcl_manager.js:399:11)] for shard shardId-000000000001
at KCLManager._handleStateInput (/usr/src/app/node_modules/aws-kcl/lib/kcl/kcl_manager.js:399:11)
Oct 22, 2018 7:47:35 PM com.amazonaws.services.kinesis.multilang.MultiLangProtocol futureMethod
SEVERE: Failed to get status message for processRecords action for shard shardId-000000000001
java.util.concurrent.ExecutionException: java.lang.RuntimeException: Reached end of STDIN of child process for shard shardId-000000000001 so won't be able to return a message.
at java.util.concurrent.FutureTask.report(FutureTask.java:122)
at java.util.concurrent.FutureTask.get(FutureTask.java:192)
at com.amazonaws.services.kinesis.multilang.MultiLangProtocol.futureMethod(MultiLangProtocol.java:197)
at com.amazonaws.services.kinesis.multilang.MultiLangProtocol.waitForStatusMessage(MultiLangProtocol.java:171)
at com.amazonaws.services.kinesis.multilang.MultiLangProtocol.waitForStatusMessage(MultiLangProtocol.java:138)
at com.amazonaws.services.kinesis.multilang.MultiLangProtocol.processRecords(MultiLangProtocol.java:92)
at com.amazonaws.services.kinesis.multilang.MultiLangRecordProcessor.processRecords(MultiLangRecordProcessor.java:108)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask.callProcessRecords(ProcessTask.java:215)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask.call(ProcessTask.java:170)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:49)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:24)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)
Caused by: java.lang.RuntimeException: Reached end of STDIN of child process for shard shardId-000000000001 so won't be able to return a message.
at com.amazonaws.services.kinesis.multilang.GetNextMessageTask.returnAfterEndOfInput(GetNextMessageTask.java:84)
at com.amazonaws.services.kinesis.multilang.GetNextMessageTask.returnAfterEndOfInput(GetNextMessageTask.java:31)
at com.amazonaws.services.kinesis.multilang.LineReaderTask.call(LineReaderTask.java:70)
... 4 more

Oct 22, 2018 7:47:35 PM com.amazonaws.services.kinesis.multilang.DrainChildSTDERRTask handleLine
SEVERE: Received error line from subprocess [ at KCLManager. (/usr/src/app/node_modules/aws-kcl/lib/kcl/kcl_manager.js:381:12)] for shard shardId-000000000001
at KCLManager. (/usr/src/app/node_modules/aws-kcl/lib/kcl/kcl_manager.js:381:12)
Oct 22, 2018 7:47:35 PM com.amazonaws.services.kinesis.multilang.MultiLangRecordProcessor stopProcessing
SEVERE: Encountered an error while trying to process records
java.lang.RuntimeException: Child process failed to process records
at com.amazonaws.services.kinesis.multilang.MultiLangRecordProcessor.processRecords(MultiLangRecordProcessor.java:109)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask.callProcessRecords(ProcessTask.java:215)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.ProcessTask.call(ProcessTask.java:170)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:49)
at com.amazonaws.services.kinesis.clientlibrary.lib.worker.MetricsCollectingTaskDecorator.call(MetricsCollectingTaskDecorator.java:24)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

---

Any help would be appreciated.

[niketh90]

Same issue here..Any help?

[brandonmino]

Any update on this?

[niketh90]

We fixed the issue. There was some issue with the braces in code and it was failing at checkpointing. After refactoring its working fine.

Here is the working copy of the code.

 `function recordProcessor () {var shardId return {
initialize: function (initializeInput, completeCallback) {
  shardId = initializeInput.shardId
  completeCallback()
},processRecords: function (processRecordsInput, completeCallback) {
  if (!processRecordsInput || !processRecordsInput.records) {
    completeCallback()
    return
  }

  var records = processRecordsInput.records
  var record, data, sequenceNumber, partitionKey
  for (var i = 0; i < records.length; ++i) {
    record = records[i]
    data = new Buffer(record.data, 'base64').toString()

         sequenceNumber = record.sequenceNumber
    partitionKey = record.partitionKey
    Logger.instance().info(util.format('ShardID: %s, Record: %s, SeqenceNumber: %s, PartitionKey:%s', shardId, data, sequenceNumber, partitionKey));
    
    ProcessMyData(data).then(result => {
         Logger.instance().info(JSON.stringify(result))
    }).catch(err => {
      Logger.instance().error('Error  ' + JSON.stringify(err))
    })
  }
  Logger.instance().info(util.format('records %s', JSON.stringify(records)))
  if (!sequenceNumber) {
    Logger.instance().info(util.format('Checkpoint not done since seq number is undefined. ShardID: %s, SeqenceNumber: %s', shardId, sequenceNumber));
    completeCallback()
    return
  } else {
    // If checkpointing, completeCallback should only be called once checkpoint is complete.
    processRecordsInput.checkpointer.checkpoint(sequenceNumber, function (_err, sequenceNumber) {
      Logger.instance().info(util.format('Checkpoint successful. ShardID: %s, SeqenceNumber: %s', shardId, sequenceNumber))
      completeCallback()
    })
  }
},
leaseLost: function (leaseLostInput, completeCallback) {
  Logger.instance().info(util.format('Lease was lost for ShardId: %s', shardId))
  completeCallback()
},

shardEnded: function (shardEndedInput, completeCallback) {
  Logger.instance().info(util.format('ShardId: %s has ended. Will checkpoint now.', shardId))
  shardEndedInput.checkpointer.checkpoint(function (_err) {
    completeCallback()
  })
},

shutdownRequested: function (shutdownRequestedInput, completeCallback) {
  shutdownRequestedInput.checkpointer.checkpoint(function (_err) {
    completeCallback()
  })
}  }}kcl(recordProcessor()).run()`

[brandonmino]

I essentially copied and pasted your code, but I still have the same error...

[vincentpham13]

Same here

[softmantk]

Did anyone find solution ?

[SimranjitKambojTR]

Seconded. We're still dealing with this issue. Did anyone find a fix? @pfifer

[ashrafuzzaman]

We are also finding the same issue and we had to restart the processor to get the consumer up and running again.

[mgagguturi]

what we have seen is the issue happens when processing time for one of the message is longer. We need to keep restarting consumer.