NIFI-15001 - Fix PutElasticsearchRecord with recursive reads #10334
Conversation
| private Record cloneRecord(final Record record) { | ||
| final Map<String, Object> valuesCopy = cloneValues(record.toMap()); | ||
| return new MapRecord(record.getSchema(), valuesCopy, record.isTypeChecked(), record.isDropUnknownFields()); | ||
| } | ||
|
|
||
| private Map<String, Object> cloneValues(final Map<String, Object> values) { | ||
| final Map<String, Object> cloned = new LinkedHashMap<>(values.size()); | ||
| for (final Map.Entry<String, Object> entry : values.entrySet()) { | ||
| cloned.put(entry.getKey(), cloneValue(entry.getValue())); | ||
| } | ||
| return cloned; | ||
| } | ||
|
|
||
| private Object cloneValue(final Object value) { |
There was a problem hiding this comment.
Instead of creating all these clone methods, it would be cleaner to use something like Apache Commons SerializationUtils#clone. The only thing you would have to make sure though is all of the data to be cloned implements the Serializable interface.
There was a problem hiding this comment.
In our case the records are MapRecord instances, and MapRecord (and other Record implementations) do not implement Serializable. Besides, Java serialization would add noticeable overhead on a hot path that can process large batches, whereas the current approach only walk the map/collection structures we actually store.
There was a problem hiding this comment.
Thanks for clarifying that. Could this cloning be reused? In other words would it make sense to have a clone/deepClone method in the Record interface and implemented in the implementations?
There was a problem hiding this comment.
I guess it could, but I'm not sure this is really a pattern we would want to encourage. The scenario here is quite specific. Others may have a different opinion though.
Summary
NIFI-15001 - Fix
PutElasticsearchRecordwith recursive readsWhen
PutElasticsearchRecordprocessed FlowFiles with batch sizes smaller than the record count, it re-read the FlowFile to build error batches. With recursive reads disabled (the default), the secondsession.read()triggeredIllegalStateException: FlowFile … already in use, so entire FlowFiles were routed to failure.The change keeps processing single-pass: every record is cloned exactly once when read from the RecordReader. The mutated copy is used for the Elasticsearch
_bulkrequest, while an untouched clone is retained for provenance and error routing. This eliminates the recursive read and keeps RecordPath mutations from leaking into error output.Tracking
Please complete the following tracking steps prior to pull request creation.
Issue Tracking
Pull Request Tracking
NIFI-00000NIFI-00000Pull Request Formatting
mainbranchVerification
Please indicate the verification steps performed prior to pull request creation.
Build
./mvnw clean install -P contrib-checkLicensing
LICENSEandNOTICEfilesDocumentation