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);
}
};
};