pvillard31 commented on code in PR #10334:
URL: https://github.com/apache/nifi/pull/10334#discussion_r2407542671
##########
nifi-extension-bundles/nifi-elasticsearch-bundle/nifi-elasticsearch-restapi-processors/src/main/java/org/apache/nifi/processors/elasticsearch/PutElasticsearchRecord.java:
##########
@@ -420,22 +423,28 @@ public void onTrigger(final ProcessContext context, final
ProcessSession session
try (final InputStream inStream = session.read(input);
final RecordReader reader =
readerFactory.createRecordReader(input, inStream, getLogger())) {
final PushBackRecordSet recordSet = new
PushBackRecordSet(reader.createRecordSet());
+ final RecordSchema recordSchema = reader.getSchema();
final List<IndexOperationRequest> operationList = new
ArrayList<>();
- final List<Record> originals = new ArrayList<>();
+ final List<Record> processedRecords = new ArrayList<>();
+ final List<Record> originalRecords = new ArrayList<>();
Record record;
while ((record = recordSet.next()) != null) {
- addOperation(operationList, record, indexOperationParameters,
indices, types);
- originals.add(record);
+ final Record originalRecord = cloneRecord(record);
+ final Record processedRecord = cloneRecord(record);
Review Comment:
Yes it is. `recordSet.next()` reuses the same Record instance each time, so
we take one deep copy before adding it to `processedRecords`. It lets us mutate
fields (for example through the various `fieldValue.updateValue(...)` calls)
without those changes leaking into later iterations. We take a second deep copy
for `originalRecords` so that, when `indexDocuments()` builds FlowFiles for
failures, it can write out an untouched version of the input rather than the
mutated version or re-read from the original FlowFile.
--
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]
