Skip to content

Commit cffe3ed

Browse files
erikayao93dreamorosiAlexander Schuerendependabot[bot]github-actions[bot]
committed
feat(batch): Implementation of base batch processing classes (#1588)
* chore: init workspace * chore: init workspace * Initial base class implementation * Added BatchProcessor implementation, attempted fix for async * Added unit tests * Refactoring unit tests * Lint fix, updated docstrings * Added response and identifier typings * test(idempotency): improve integration tests for utility (#1591) * docs: new name * chore: rename e2e files * tests(idempotency): expand integration tests * chore(idempotency): remove unreachable code * Removed unnecessary type casting * Moved exports for handlers and factories * Updated imports, refactored randomization in factories * Refactored EventType to be const instead of enum * Refactored and added documentation for errors * Removed debugging line * chore(ci): add canary to layer deployment (#1593) * docs(idempotency): write utility docs (#1592) * docs: base docs * wip * chore: added paths to snippets tsconfig * chore: added page to docs menu * docs(idempotency): utility docs * highlights * chore: remove CDK mention * build(internal): bump semver from 5.7.1 to 5.7.2 (#1594) Bumps [semver](https://github.com/npm/node-semver) from 5.7.1 to 5.7.2. - [Release notes](https://github.com/npm/node-semver/releases) - [Changelog](https://github.com/npm/node-semver/blob/v5.7.2/CHANGELOG.md) - [Commits](npm/node-semver@v5.7.1...v5.7.2) --- updated-dependencies: - dependency-name: semver dependency-type: indirect ... Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> * chore(idempotency): mark the utility ready public beta (#1595) * chore(idempotency): mark utility as public beta * chore: manually increment version in commons * docs(internal): update AWS SDK links to new docs (#1597) * chore(maintenance): remove parameters utility from layer bundling and layers e2e tests (#1599) * remove parameter from e2e tests * remove parameters from canary stack as well * chore(release): v1.11.1 [skip ci] * fix canary deploy in ci with correct workspace name (#1601) * chore: update layer ARN on documentation --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: Andrea Amorosi <dreamorosi@gmail.com> Co-authored-by: Alexander Schueren <amelnyk@amazon.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: Release bot[bot] <aws-devax-open-source@amazon.com>
1 parent 5764ee8 commit cffe3ed

11 files changed

+969
-8
lines changed

‎package-lock.json

+5-1
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎packages/batch/src/BasePartialBatchProcessor.ts

+161
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,161 @@
1+
/**
2+
* Process batch and partially report failed items
3+
*/
4+
import { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from 'aws-lambda';
5+
import {
6+
BasePartialProcessor,
7+
BatchProcessingError,
8+
DATA_CLASS_MAPPING,
9+
DEFAULT_RESPONSE,
10+
EventSourceDataClassTypes,
11+
EventType,
12+
ItemIdentifier,
13+
BatchResponse,
14+
} from '.';
15+
16+
abstract class BasePartialBatchProcessor extends BasePartialProcessor {
17+
public COLLECTOR_MAPPING;
18+
19+
public batchResponse: BatchResponse;
20+
21+
public eventType: keyof typeof EventType;
22+
23+
/**
24+
* Initializes base batch processing class
25+
* @param eventType Whether this is SQS, DynamoDB stream, or Kinesis data stream event
26+
*/
27+
public constructor(eventType: keyof typeof EventType) {
28+
super();
29+
this.eventType = eventType;
30+
this.batchResponse = DEFAULT_RESPONSE;
31+
this.COLLECTOR_MAPPING = {
32+
[EventType.SQS]: () => this.collectSqsFailures(),
33+
[EventType.KinesisDataStreams]: () => this.collectKinesisFailures(),
34+
[EventType.DynamoDBStreams]: () => this.collectDynamoDBFailures(),
35+
};
36+
}
37+
38+
/**
39+
* Report messages to be deleted in case of partial failures
40+
*/
41+
public clean(): void {
42+
if (!this.hasMessagesToReport()) {
43+
return;
44+
}
45+
46+
if (this.entireBatchFailed()) {
47+
throw new BatchProcessingError(
48+
'All records failed processing. ' +
49+
this.exceptions.length +
50+
' individual errors logged separately below.',
51+
this.exceptions
52+
);
53+
}
54+
55+
const messages: ItemIdentifier[] = this.getMessagesToReport();
56+
this.batchResponse = { batchItemFailures: messages };
57+
}
58+
59+
/**
60+
* Collects identifiers of failed items for a DynamoDB stream
61+
* @returns list of identifiers for failed items
62+
*/
63+
public collectDynamoDBFailures(): ItemIdentifier[] {
64+
const failures: ItemIdentifier[] = [];
65+
66+
for (const msg of this.failureMessages) {
67+
const msgId = (msg as DynamoDBRecord).dynamodb?.SequenceNumber;
68+
if (msgId) {
69+
failures.push({ itemIdentifier: msgId });
70+
}
71+
}
72+
73+
return failures;
74+
}
75+
76+
/**
77+
* Collects identifiers of failed items for a Kinesis stream
78+
* @returns list of identifiers for failed items
79+
*/
80+
public collectKinesisFailures(): ItemIdentifier[] {
81+
const failures: ItemIdentifier[] = [];
82+
83+
for (const msg of this.failureMessages) {
84+
const msgId = (msg as KinesisStreamRecord).kinesis.sequenceNumber;
85+
failures.push({ itemIdentifier: msgId });
86+
}
87+
88+
return failures;
89+
}
90+
91+
/**
92+
* Collects identifiers of failed items for an SQS batch
93+
* @returns list of identifiers for failed items
94+
*/
95+
public collectSqsFailures(): ItemIdentifier[] {
96+
const failures: ItemIdentifier[] = [];
97+
98+
for (const msg of this.failureMessages) {
99+
const msgId = (msg as SQSRecord).messageId;
100+
failures.push({ itemIdentifier: msgId });
101+
}
102+
103+
return failures;
104+
}
105+
106+
/**
107+
* Determines whether all records in a batch failed to process
108+
* @returns true if all records resulted in exception results
109+
*/
110+
public entireBatchFailed(): boolean {
111+
return this.exceptions.length == this.records.length;
112+
}
113+
114+
/**
115+
* Collects identifiers for failed batch items
116+
* @returns formatted messages to use in batch deletion
117+
*/
118+
public getMessagesToReport(): ItemIdentifier[] {
119+
return this.COLLECTOR_MAPPING[this.eventType]();
120+
}
121+
122+
/**
123+
* Determines if any records failed to process
124+
* @returns true if any records resulted in exception
125+
*/
126+
public hasMessagesToReport(): boolean {
127+
if (this.failureMessages.length != 0) {
128+
return true;
129+
}
130+
131+
// console.debug('All ' + this.successMessages.length + ' records successfully processed');
132+
133+
return false;
134+
}
135+
136+
/**
137+
* Remove results from previous execution
138+
*/
139+
public prepare(): void {
140+
this.successMessages.length = 0;
141+
this.failureMessages.length = 0;
142+
this.exceptions.length = 0;
143+
this.batchResponse = DEFAULT_RESPONSE;
144+
}
145+
146+
/**
147+
* @returns Batch items that failed processing, if any
148+
*/
149+
public response(): BatchResponse {
150+
return this.batchResponse;
151+
}
152+
153+
public toBatchType(
154+
record: EventSourceDataClassTypes,
155+
eventType: keyof typeof EventType
156+
): SQSRecord | KinesisStreamRecord | DynamoDBRecord {
157+
return DATA_CLASS_MAPPING[eventType](record);
158+
}
159+
}
160+
161+
export { BasePartialBatchProcessor };

‎packages/batch/src/BasePartialProcessor.ts

+120
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,120 @@
1+
/**
2+
* Abstract class for batch processors
3+
*/
4+
import {
5+
BaseRecord,
6+
EventSourceDataClassTypes,
7+
FailureResponse,
8+
ResultType,
9+
SuccessResponse,
10+
} from '.';
11+
12+
abstract class BasePartialProcessor {
13+
public exceptions: Error[];
14+
15+
public failureMessages: EventSourceDataClassTypes[];
16+
17+
public handler: CallableFunction;
18+
19+
public records: BaseRecord[];
20+
21+
public successMessages: EventSourceDataClassTypes[];
22+
23+
/**
24+
* Initializes base processor class
25+
*/
26+
public constructor() {
27+
this.successMessages = [];
28+
this.failureMessages = [];
29+
this.exceptions = [];
30+
this.records = [];
31+
this.handler = new Function();
32+
}
33+
34+
/**
35+
* Clean class instance after processing
36+
*/
37+
public abstract clean(): void;
38+
39+
/**
40+
* Keeps track of batch records that failed processing
41+
* @param record record that failed processing
42+
* @param exception exception that was thrown
43+
* @returns FailureResponse object with ["fail", exception, original record]
44+
*/
45+
public failureHandler(
46+
record: EventSourceDataClassTypes,
47+
exception: Error
48+
): FailureResponse {
49+
const entry: FailureResponse = ['fail', exception.message, record];
50+
// console.debug('Record processing exception: ' + exception.message);
51+
this.exceptions.push(exception);
52+
this.failureMessages.push(record);
53+
54+
return entry;
55+
}
56+
57+
/**
58+
* Prepare class instance before processing
59+
*/
60+
public abstract prepare(): void;
61+
62+
/**
63+
* Call instance's handler for each record
64+
* @returns List of processed records
65+
*/
66+
public async process(): Promise<(SuccessResponse | FailureResponse)[]> {
67+
this.prepare();
68+
69+
const processedRecords: (SuccessResponse | FailureResponse)[] = [];
70+
for (const record of this.records) {
71+
processedRecords.push(await this.processRecord(record));
72+
}
73+
74+
this.clean();
75+
76+
return processedRecords;
77+
}
78+
79+
/**
80+
* Process a record with the handler
81+
* @param record Record to be processed
82+
*/
83+
public abstract processRecord(
84+
record: BaseRecord
85+
): Promise<SuccessResponse | FailureResponse>;
86+
87+
/**
88+
* Set class instance attributes before execution
89+
* @param records List of records to be processed
90+
* @param handler CallableFunction to process entries of "records"
91+
* @returns this object
92+
*/
93+
public register(
94+
records: BaseRecord[],
95+
handler: CallableFunction
96+
): BasePartialProcessor {
97+
this.records = records;
98+
this.handler = handler;
99+
100+
return this;
101+
}
102+
103+
/**
104+
* Keeps track of batch records that were processed successfully
105+
* @param record record that succeeded processing
106+
* @param result result from record handler
107+
* @returns SuccessResponse object with ["success", result, original record]
108+
*/
109+
public successHandler(
110+
record: EventSourceDataClassTypes,
111+
result: ResultType
112+
): SuccessResponse {
113+
const entry: SuccessResponse = ['success', result, record];
114+
this.successMessages.push(record);
115+
116+
return entry;
117+
}
118+
}
119+
120+
export { BasePartialProcessor };

‎packages/batch/src/BatchProcessor.ts

+29-1
Original file line numberDiff line numberDiff line change
@@ -1,3 +1,31 @@
1-
class BatchProcessor {}
1+
/**
2+
* Process native partial responses from SQS, Kinesis Data Streams, and DynamoDB
3+
*/
4+
import {
5+
BasePartialBatchProcessor,
6+
BaseRecord,
7+
FailureResponse,
8+
SuccessResponse,
9+
} from '.';
10+
11+
class BatchProcessor extends BasePartialBatchProcessor {
12+
/**
13+
* Process a record with instance's handler
14+
* @param record Batch record to be processed
15+
* @returns response of success or failure
16+
*/
17+
public async processRecord(
18+
record: BaseRecord
19+
): Promise<SuccessResponse | FailureResponse> {
20+
try {
21+
const data = this.toBatchType(record, this.eventType);
22+
const result = await this.handler(data);
23+
24+
return this.successHandler(record, result);
25+
} catch (e) {
26+
return this.failureHandler(record, e as Error);
27+
}
28+
}
29+
}
230

331
export { BatchProcessor };

‎packages/batch/src/constants.ts

+25
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,25 @@
1+
/**
2+
* Constants for batch processor classes
3+
*/
4+
import { DynamoDBRecord, KinesisStreamRecord, SQSRecord } from 'aws-lambda';
5+
import type { BatchResponse, EventSourceDataClassTypes } from '.';
6+
7+
const EventType = {
8+
SQS: 'SQS',
9+
KinesisDataStreams: 'KinesisDataStreams',
10+
DynamoDBStreams: 'DynamoDBStreams',
11+
} as const;
12+
13+
const DEFAULT_RESPONSE: BatchResponse = {
14+
batchItemFailures: [],
15+
};
16+
17+
const DATA_CLASS_MAPPING = {
18+
[EventType.SQS]: (record: EventSourceDataClassTypes) => record as SQSRecord,
19+
[EventType.KinesisDataStreams]: (record: EventSourceDataClassTypes) =>
20+
record as KinesisStreamRecord,
21+
[EventType.DynamoDBStreams]: (record: EventSourceDataClassTypes) =>
22+
record as DynamoDBRecord,
23+
};
24+
25+
export { EventType, DEFAULT_RESPONSE, DATA_CLASS_MAPPING };

0 commit comments

Comments
 (0)