[FLINK-35989] Document and log error on partially failed requests for AWS sinks

• Type: Improvement
• Status: Open
• Priority: Major
• Resolution: Unresolved
• Affects Version/s: aws-connector-4.3.0
• Fix Version/s: aws-connector-4.4.0
• Component/s: Connectors / DynamoDB, Connectors / Firehose, Connectors / Kinesis
• Assignee: Unassigned
• Reporter: Hong Liang Teoh
• Created: 06/Aug/24 09:34
• Updated: 04/Nov/24 22:16

Description

Problem

Currently, AWS sinks send batch write requests to endpoints.

On partial failure within the batch (only some records fail), we perform retries or fail the job, but we don't log the error returned.

    private void handlePartiallyFailedRequest(
            PutRecordsResponse response,
            List<PutRecordsRequestEntry> requestEntries,
            Consumer<List<PutRecordsRequestEntry>> requestResult) {
        LOG.warn(
                "KDS Sink failed to write and will retry {} entries to KDS",
                response.failedRecordCount());
        numRecordsOutErrorsCounter.inc(response.failedRecordCount());
        if (failOnError) {
            getFatalExceptionCons()
                    .accept(new KinesisStreamsException.KinesisStreamsFailFastException());
            return;
        }
        List<PutRecordsRequestEntry> failedRequestEntries =
                new ArrayList<>(response.failedRecordCount());
        List<PutRecordsResultEntry> records = response.records();
        for (int i = 0; i < records.size(); i++) {
            if (records.get(i).errorCode() != null) {
                failedRequestEntries.add(requestEntries.get(i));
            }
        }
        requestResult.accept(failedRequestEntries);
    }

Code link: https://github.com/apache/flink-connector-aws/blob/main/flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkWriter.java#L254-L274

What we want to do

We should log the reason for failure to assist user debugging.