diff --git a/lib/stream/leo-stream.js b/lib/stream/leo-stream.js index d06e02d..69b6c3b 100644 --- a/lib/stream/leo-stream.js +++ b/lib/stream/leo-stream.js @@ -884,15 +884,8 @@ module.exports = function(configure) { logger.error("toLeo.firehose.putRecordBatch", err); retry.backoff(); } else if (data.FailedPutCount && data.FailedPutCount > 0) { - var left = []; - for (var i = 0; i < data.RequestResponses.length; i++) { - var row = data.RequestResponses[i]; - if (row.ErrorCode) { - left.push(records[i]); - } - } - logger.info(`Count failed firehose put record batch (failed, retry):(${data.FailedPutCount},${left.length})`); - records = left; + let response = (data.RequestResponses && data.RequestResponses[0]) || {}; + logger.info(`Count failed firehose put record batch (failed records, buffered chunks):(${data.FailedPutCount},${records.length})`, response.ErrorCode, response.ErrorMessage); retry.backoff(); } else { logger.timeEnd("firehose request" + loggerTimeId); diff --git a/package-lock.json b/package-lock.json index 4bb0253..1c656f5 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "leo-sdk", - "version": "7.1.16", + "version": "7.1.17", "lockfileVersion": 2, "requires": true, "packages": { "": { "name": "leo-sdk", - "version": "7.1.16", + "version": "7.1.17", "license": "MIT", "dependencies": { "async": "2.6.4", diff --git a/package.json b/package.json index 461a9bd..4dd464a 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "leo-sdk", - "version": "7.1.16", + "version": "7.1.17", "description": "Load data onto the LEO Platform", "homepage": "https://leoplatform.io", "main": "index.js", diff --git a/test/lib.stream.leo-stream.write.utest.ts b/test/lib.stream.leo-stream.write.utest.ts index cd8da77..875b8d2 100644 --- a/test/lib.stream.leo-stream.write.utest.ts +++ b/test/lib.stream.leo-stream.write.utest.ts @@ -191,10 +191,27 @@ describe("lib/stream/leo-stream.js write", function () { assert.deepEqual(events.map((e) => e.payload), [{ n: 1 }, { n: 2 }]); }); - it("firehose retries only the failed records of a partial failure", async function () { + it("firehose resends the whole combined record when it is rejected", async function () { putRecordBatch.onFirstCall().callsFake((p: any, cb: any) => cb(null, { FailedPutCount: 1, RequestResponses: [{ ErrorCode: "ServiceUnavailableException" }] })); - await write([evt(1)], ls.toLeo("bot1", { ...fh, backoff: { initialDelay: 1, maxDelay: 2 } })); + await write([evt(1), evt(2), evt(3)], ls.toLeo("bot1", { ...fh, gzipFirehose: true, records: 10, chunk: { records: 1 }, backoff: { initialDelay: 1, maxDelay: 2 } })); + assert.equal(putRecordBatch.callCount, 2); + const first = putRecordBatch.getCall(0).args[0].Records[0].Data; + const second = putRecordBatch.getCall(1).args[0].Records[0].Data; + assert.isAtLeast(first.toString().trim().split("\n").length, 2); + assert.isTrue(Buffer.from(second).equals(Buffer.from(first))); + const resent = second.toString().trim().split("\n").flatMap((l: string) => decodeGzipRecord(Buffer.from(l, "base64"))); + assert.deepEqual(resent.map((e) => e.payload), [{ n: 1 }, { n: 2 }, { n: 3 }]); + }); + + it("firehose resends the whole raw record when it is rejected", async function () { + putRecordBatch.onFirstCall().callsFake((p: any, cb: any) => cb(null, { FailedPutCount: 1, RequestResponses: [{ ErrorCode: "ServiceUnavailableException" }] })); + await write([evt(1), evt(2), evt(3)], ls.toLeo("bot1", { ...fh, records: 10, chunk: { records: 1 }, backoff: { initialDelay: 1, maxDelay: 2 } })); assert.equal(putRecordBatch.callCount, 2); + const first = putRecordBatch.getCall(0).args[0].Records[0].Data; + const second = putRecordBatch.getCall(1).args[0].Records[0].Data; + assert.isTrue(Buffer.from(second).equals(Buffer.from(first))); + const resent = second.toString().trim().split("\n").map((l: string) => JSON.parse(l)); + assert.deepEqual(resent.map((e: any) => e.payload), [{ n: 1 }, { n: 2 }, { n: 3 }]); }); it("firehose retries after an error", async function () {