[
https://issues.apache.org/jira/browse/FLINK-35989?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Hong Liang Teoh updated FLINK-35989:
------------------------------------
Description:
h2. 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.
{code:java}
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}
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]
h2. What we want to do
We should log the reason for failure to assist user debugging.
was:
h2. 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.
{code:java}
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}
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]
h2. What we want to do
We should log the reason for failure to assist user debugging.
> Document and log error on partially failed requests for AWS sinks
> -----------------------------------------------------------------
>
> Key: FLINK-35989
> URL: https://issues.apache.org/jira/browse/FLINK-35989
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / DynamoDB, Connectors / Firehose, Connectors
> / Kinesis
> Affects Versions: aws-connector-4.3.0
> Reporter: Hong Liang Teoh
> Priority: Major
> Fix For: aws-connector-4.4.0
>
>
> h2. 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.
> {code:java}
> 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}
> 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]
> h2. What we want to do
> We should log the reason for failure to assist user debugging.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
