[GitHub] [nifi] timeabarna commented on a change in pull request #5593: NIFI-9475 Provide Framework-Level Retries for NiFi Relationships

[GitBox <gi...@apache.org>]

timeabarna commented on a change in pull request #5593:
URL: https://github.com/apache/nifi/pull/5593#discussion_r766673035



##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
##########
@@ -356,6 +408,95 @@ private void checkpoint(final boolean copyCollections) {
         checkpoint.checkpoint(this, autoTerminatedEvents, copyCollections);
     }
 
+    private boolean isRetryNeeded(final ProcessorNode processorNode, final StandardRepositoryRecord record, final FlowFileRecord currentFlowFile,
+                                  final int retryCounts, final Map<String, StandardRepositoryRecord> uuidsToRecords) {
+        if (currentFlowFile == null || processorNode == null || processorNode.getRetriedRelationships().isEmpty()) {
+            return false;
+        }
+
+        if (processorNode.isRetriedRelationship(record.getTransferRelationship())) {
+            return retryCounts < processorNode.getRetryCounts();
+        }
+
+        if (forkEventBuilders.get(currentFlowFile) != null) {
+            for (String uuid : forkEventBuilders.get(currentFlowFile).getChildFlowFileIds()) {
+                if (processorNode.isRetriedRelationship(uuidsToRecords.get(uuid).getTransferRelationship())) {
+                    return retryCounts < processorNode.getRetryCounts();
+                }
+            }
+        }
+        return false;
+    }
+
+    private FlowFileRecord updateFlowFileRecord(final StandardRepositoryRecord record,
+                                                final Map<String, StandardRepositoryRecord> uuidsToRecords,
+                                                final int retryCounts, final FlowFileRecord flowFileRecord) {
+
+        removeTemporaryClaim(record);
+        if (forkEventBuilders.get(flowFileRecord) != null) {
+            for (String uuid : forkEventBuilders.get(flowFileRecord).getChildFlowFileIds()) {
+                final StandardRepositoryRecord childRecord = uuidsToRecords.get(uuid);

Review comment:
       Using the uuid map at different places to get records based on their uuid otherwise I would need another loop through records.values to identify the record.

##########
File path: nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java
##########
@@ -289,63 +292,112 @@ private void checkpoint(final boolean copyCollections) {
 
         // validate that all records have a transfer relationship for them and if so determine the destination node and clone as necessary
         final Map<Long, StandardRepositoryRecord> toAdd = new HashMap<>();
+        final Map<String, StandardRepositoryRecord> uuidsToRecords = records.values()

Review comment:
       I create a map for records to be accessible by their workingflowfile uuids




-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org