[improve][es-sink] Add error log for failed bulk records in ES sink
@@ -76,7 +76,7 @@ public class ElasticSearchClient implements AutoCloseable {
final Record record = bulkOperationList.get(index++).getPulsarRecord();
if (result.isError()) {
record.fail();
- checkForIrrecoverableError(result);
+ checkForIrrecoverableError(record, result);
} else {
record.ack();
}
@@ -106,13 +106,15 @@ public class ElasticSearchClient implements AutoCloseable {
return irrecoverableError.get() != null;
}
- void checkForIrrecoverableError(BulkProcessor.BulkOperationResult result) {
+ void checkForIrrecoverableError(Record<?> record, BulkProcessor.BulkOperationResult result) {
if (!result.isError()) {
return;
}
final String errorCause = result.getError();
+ boolean isMalformed = false;
for (String error : MALFORMED_ERRORS) {
if (errorCause.contains(error)) {
+ isMalformed = true;
switch (config.getMalformedDocAction()) {
case IGNORE:
break;
@@ -132,6 +134,13 @@ public class ElasticSearchClient implements AutoCloseable {
}
}
}
+ if (!isMalformed) {
+ log.warn("Bulk request failed, message id=[{}] index={} error={}",
+ record.getMessage()
+ .map(m -> m.getMessageId().toString())
+ .orElse(""),
+ result.getIndex(), result.getError());
+ }
}
public void bulkIndex(Record record, Pair<String, String> idAndDoc) throws Exception {
