aws-nageshvh commented on code in PR #201:
URL:
https://github.com/apache/flink-connector-aws/pull/201#discussion_r2274833916
##########
flink-connector-aws/flink-connector-aws-kinesis-streams/src/main/java/org/apache/flink/connector/kinesis/sink/KinesisStreamsSinkWriter.java:
##########
@@ -244,34 +314,121 @@ private void handleFullyFailedRequest(
@Override
public void close() {
- AWSGeneralUtil.closeResources(httpClient, kinesisClient);
+ try {
+ kinesisClientProvider.close();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to close the
kinesisClientProvider", e);
+ }
}
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());
+ int failedRecordCount = response.failedRecordCount();
+ LOG.warn("KDS Sink failed to write and will retry {} entries to KDS",
failedRecordCount);
+ numRecordsOutErrorsCounter.inc(failedRecordCount);
if (failOnError) {
getFatalExceptionCons()
.accept(new
KinesisStreamsException.KinesisStreamsFailFastException());
return;
}
- List<PutRecordsRequestEntry> failedRequestEntries =
- new ArrayList<>(response.failedRecordCount());
+
+ List<PutRecordsRequestEntry> failedRequestEntries = new
ArrayList<>(failedRecordCount);
List<PutRecordsResultEntry> records = response.records();
+ // Collect error information and build the list of failed entries
+ Map<String, ErrorSummary> errorSummaries =
+ collectErrorSummaries(records, requestEntries,
failedRequestEntries);
+
+ // Log aggregated error information
+ logErrorSummaries(errorSummaries);
+
+ requestResult.accept(failedRequestEntries);
+ }
+
+ /**
+ * Collect error summaries from failed records and build a list of failed
request entries.
+ *
+ * @param records The result entries from the Kinesis response
+ * @param requestEntries The original request entries
+ * @param failedRequestEntries List to populate with failed entries
(modified as a side effect)
+ * @return A map of error codes to their summaries
+ */
+ private Map<String, ErrorSummary> collectErrorSummaries(
+ List<PutRecordsResultEntry> records,
+ List<PutRecordsRequestEntry> requestEntries,
+ List<PutRecordsRequestEntry> failedRequestEntries) {
+
+ // We capture error info while minimizing logging overhead in the data
path,
+ // which is critical for maintaining throughput performance
+ Map<String, ErrorSummary> errorSummaries = new HashMap<>();
+
for (int i = 0; i < records.size(); i++) {
- if (records.get(i).errorCode() != null) {
+ PutRecordsResultEntry resultEntry = records.get(i);
+ String errorCode = resultEntry.errorCode();
+
+ if (errorCode != null) {
+ // Track the frequency of each error code to identify patterns
+ ErrorSummary summary =
+ errorSummaries.computeIfAbsent(
+ errorCode, code -> new
ErrorSummary(resultEntry.errorMessage()));
+ summary.incrementCount();
+
failedRequestEntries.add(requestEntries.get(i));
}
}
- requestResult.accept(failedRequestEntries);
+ return errorSummaries;
+ }
+
+ /**
+ * Log aggregated error information at WARN level.
+ *
+ * @param errorSummaries Map of error codes to their summaries
+ */
+ private void logErrorSummaries(Map<String, ErrorSummary> errorSummaries) {
+ // We log aggregated error information at WARN level to ensure
visibility in production
+ // while avoiding the performance impact of logging each individual
failure
+ if (!errorSummaries.isEmpty()) {
+ StringBuilder errorSummary = new StringBuilder("Kinesis errors
summary: ");
+ errorSummaries.forEach(
+ (code, summary) ->
+ errorSummary.append(
+ String.format(
+ "[%s: %d records, example: %s] ",
+ code,
+ summary.getCount(),
+ summary.getExampleMessage())));
Review Comment:
makes sense. was in two minds because the format looked off but ended up
changing that for better encapsulation
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
