1
\$\begingroup\$

My application is really easy. It's going to add 500 records at a time to AWS Kinesis and if there's an error occur I'll just add that back to the queue and it's going to be retry. It's going well but I feel like there's a bug somewhere that I don't see.

'use strict';

const _ = require('lodash');
const moment = require('moment');

const config = require('../config');
const logger = require('../logger');
const numbers = require('../helpers/numbers');

const streamName = config.kinesis.streamName;

let records = [];
module.exports = (kinesis) => {
  let sendRecords = () => {
    let payloadRecords = {
      Records: records,
      StreamName: streamName
    };

    if (records.length >= 500) {
      const pushingToKinesis = records.splice(0, 500);
      payloadRecords.Records = pushingToKinesis;
    } else {
      records = []
    }

    kinesis.putRecords(payloadRecords, (err, data) => {
      if (err) {
        logger.error(err);
      }

      const failedRecord = _.get(data, 'FailedRecordCount', 0);

      if (failedRecord > 0) {

        logger.warn(`There are ${data.FailedRecordCount} failed`);

        data.Records.forEach((record, index) => {
          if (_.has(record, 'ErrorCode') || _.has(record, 'ErrorMessage')) {
            logger.debug(record);
            logger.debug(payloadRecords.Records[index]);

            records.push(payloadRecords.Records[index]);
          }
        });
      }
    });
  };

  let putRecord = (record) => {
    let payload = {
      Data: JSON.stringify(record),
      PartitionKey: String(numbers.random() * 100000)
    };
    records.push(payload);

    if (records.length >= config.kinesis.maxRecords) {
      sendRecords();
    }
  };

  return {
    putRecord: putRecord,
    recordCount: () => {
      return records.length;
    },
    clearRecord: () => {
      records = [];
    },
    init: () => {
      setInterval(() => {
        if (!_.isEmpty(records)) { 
          sendRecords();
        }
      }, 200);
    }
  };
};
\$\endgroup\$
  • 3
    \$\begingroup\$ "I feel like there's a bug somewhere I don't see." - Do you mean that you're worried about overlooked issues or are you actually getting bad data from this script? \$\endgroup\$ – SuperBiasedMan Jul 5 '16 at 13:34
  • \$\begingroup\$ Yes. I'm worried that I might have overlooked something. Am I in the wrong place? \$\endgroup\$ – toy Jul 5 '16 at 18:37
1
\$\begingroup\$

There is no obvious bug but;

  • You are defining functions both within the return structure and outside of it, which makes it hard to read/parse, just keep it clean and define everything outside of it
  • You are sending records in a function called putRecord, that is not good.
  • 500 should probably be a constant retrieved with config, same thing for 200
  • Using numbers.random() sounds like a terrible idea, you should use a library that generates GUID's (https://www.npmjs.com/package/guid)
  • Why assign config.kinesis.streamName to streamName, just assign it straight in the structure. It is one less line of code, and the reader does not have to wonder where it comes from
  • I would have re-ordered the 500 check and the assignment:

    records  = records.splice(0, 500);
    
    let payloadRecords = {
      Records: records,
      StreamName: streamName
    };
    

    instead of

    let payloadRecords = {
      Records: records,
     StreamName: streamName
    };
    
    if (records.length >= 500) {
      const pushingToKinesis = records.splice(0, 500);
      payloadRecords.Records = pushingToKinesis;
    } else {
      records = []
    }
    

    personally I would even go for

    //Make sure we never send more than 500 records, splice keeps the rest in records
    let payloadRecords = {
      Records: records.splice(0, 500),
      StreamName: streamName
    };
    

Finally, I am not a big fan of perpetually re-sending the data. If the data gets rejected because of data-content, then at some point if you reach 500 entries that will never be stored, your pipe will filled with junk and you will stop transmitting store-able data.

\$\endgroup\$
  • \$\begingroup\$ Thanks for this. It really is helpful. Right now, I have no way of checking bad data because Kinesis only returns two things; ProvisionThroughPut and InternalFailure and they can be thrown anytime according to AWS documentation. \$\endgroup\$ – toy Jul 6 '16 at 21:44
  • 1
    \$\begingroup\$ You could keep track of the re-tries in a separate structure (GUID/retrycount), and remove from the queue after 10 tries. \$\endgroup\$ – konijn Jul 7 '16 at 15:15

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.