MartijnVisser commented on code in PR #107:
URL:
https://github.com/apache/flink-connector-elasticsearch/pull/107#discussion_r4360273566
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriter.java:
##########
@@ -80,6 +84,9 @@ public class Elasticsearch8AsyncWriter<InputT> extends
AsyncSinkWriter<InputT, O
"Could not connect to
Elasticsearch cluster using the provided hosts",
err)));
+ private static final Set<Integer> ELASTICSEARCH_NON_RETRYABLE_STATUS =
Review Comment:
A 409 for an index with `ifSeqNo`, an external version or op type `create`
never succeeds on retry, only update conflicts can. Replaying one externally
versioned record on this branch gave about 500 bulk requests per second and
`flush(false)` never returned, so checkpoints hang. I think 409 should only be
retried for updates.
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriter.java:
##########
@@ -160,21 +167,45 @@ private void handlePartiallyFailedRequest(
ResultHandler<Operation> resultHandler,
BulkResponse response) {
LOG.debug("The BulkRequest has failed partially. Response: {}",
response);
- ArrayList<Operation> failedItems = new ArrayList<>();
+
+ ArrayList<Operation> failedItemsToRetry = new ArrayList<>();
+ int totalFailedItems = 0;
+ FlinkRuntimeException nonRetryableException = null;
+
for (int i = 0; i < response.items().size(); i++) {
- if (response.items().get(i).error() != null) {
- failedItems.add(requestEntries.get(i));
+ BulkResponseItem responseItem = response.items().get(i);
+ if (responseItem.error() != null) {
+ totalFailedItems++;
+ if (isOperationRetryable(responseItem.status())) {
+ failedItemsToRetry.add(requestEntries.get(i));
+ } else {
+ LOG.error(
+ "Failed to process non-retryable operation: {},
response: {}",
+ requestEntries.get(i),
+ responseItem);
+ nonRetryableException =
+ new FlinkRuntimeException(
+ "Failed to process non-retryable
operation, reason=%s"
Review Comment:
`FlinkRuntimeException` doesn't format, so this reads `reason=%s[missing-1]:
document missing`. Please add status, error type, index and id, it's all the
user gets.
##########
flink-connector-elasticsearch8/src/main/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriter.java:
##########
@@ -160,21 +167,45 @@ private void handlePartiallyFailedRequest(
ResultHandler<Operation> resultHandler,
BulkResponse response) {
LOG.debug("The BulkRequest has failed partially. Response: {}",
response);
- ArrayList<Operation> failedItems = new ArrayList<>();
+
+ ArrayList<Operation> failedItemsToRetry = new ArrayList<>();
+ int totalFailedItems = 0;
+ FlinkRuntimeException nonRetryableException = null;
+
for (int i = 0; i < response.items().size(); i++) {
- if (response.items().get(i).error() != null) {
- failedItems.add(requestEntries.get(i));
+ BulkResponseItem responseItem = response.items().get(i);
+ if (responseItem.error() != null) {
+ totalFailedItems++;
+ if (isOperationRetryable(responseItem.status())) {
+ failedItemsToRetry.add(requestEntries.get(i));
+ } else {
+ LOG.error(
+ "Failed to process non-retryable operation: {},
response: {}",
+ requestEntries.get(i),
+ responseItem);
+ nonRetryableException =
+ new FlinkRuntimeException(
+ "Failed to process non-retryable
operation, reason=%s"
+ + responseItem.error().reason());
+ break;
Review Comment:
After the `break`, `numRecordsOutErrors` only counts up to the first
non-retryable item (1 of 3 in my run). Count all of them and chain the rest as
suppressed?
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java:
##########
@@ -195,7 +249,9 @@ public void testHandlePartiallyFailedBulk() throws
Exception {
await();
+ // Verify that non-retryable error (404) increments error counter and
fails fast
assertThat(context.metricGroup().getNumRecordsOutErrorsCounter().getCount()).isEqualTo(1);
+ assertThat(completedExceptionally.get()).isTrue();
Review Comment:
Might make sense to assert the message here too, that would have caught the
`%s`.
##########
flink-connector-elasticsearch8/src/test/java/org/apache/flink/connector/elasticsearch/sink/Elasticsearch8AsyncWriterITCase.java:
##########
@@ -171,8 +174,59 @@ public void testSendTimeMetric() throws Exception {
@Timeout(5)
public void testHandlePartiallyFailedBulk() throws Exception {
String index = "test-partially-failed-bulk";
+ int maxBatchSize = 3;
+
+ // First create a document to enable version conflict
+ try (final Elasticsearch8AsyncWriter<DummyData> setupWriter =
createWriter(index, 1)) {
+ setupWriter.write(new DummyData("test-3", "test-3"), null);
+ await();
+ }
+
+ // Create converter that triggers 409 version conflict for test-3
+ Elasticsearch8AsyncSinkBuilder.OperationConverter<DummyData>
conflictConverter =
+ new Elasticsearch8AsyncSinkBuilder.OperationConverter<>(
+ (element, ctx) -> {
+ if (element.getId().equals("test-3")) {
+ // Use wrong version to trigger 409 conflict
(retryable)
+ return new IndexOperation.Builder<>()
+ .id(element.getId())
+ .index(index)
+ .document(element)
+ .ifSeqNo(999L) // Wrong sequence number
+ .ifPrimaryTerm(1L)
+ .build();
+ } else {
+ return new IndexOperation.Builder<>()
+ .id(element.getId())
+ .index(index)
+ .document(element)
+ .build();
+ }
+ });
+
+ try (final Elasticsearch8AsyncWriter<DummyData> writer =
+ createWriter(maxBatchSize, conflictConverter)) {
+ writer.write(new DummyData("test-1", "test-1"), null);
+ writer.write(new DummyData("test-2", "test-2"), null);
+ writer.write(new DummyData("test-3", "version-conflict"), null);
+ }
+
+ await();
+
+ // 409 is retryable, so test-3 should have not completed the rest
handler exceptionally
Review Comment:
This pins a retry that can never succeed. Once 409 is settled this should
assert that outcome.
--
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]
