NIFI-15904 Fixed Lineage Start Index in Session.create() (apache#11205)

- Set explicit Lineage Start Index using FlowFile ID
- Set explicit Entry Date and Lineage Start Date on new FlowFiles

Author: exceptionfactory

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/main/java/org/apache/nifi/controller/repository/StandardProcessSession.java

@@ -2069,8 +2069,13 @@ public FlowFile create() {
         attrs.put(CoreAttributes.PATH.key(), DEFAULT_FLOWFILE_PATH);
         attrs.put(CoreAttributes.UUID.key(), uuid);
 
-        final FlowFileRecord fFile = new StandardFlowFileRecord.Builder().id(context.getNextFlowFileSequence())
+        final long entryDate = System.currentTimeMillis();
+        final long id = context.getNextFlowFileSequence();
+        final FlowFileRecord fFile = new StandardFlowFileRecord.Builder().id(id)
             .addAttributes(attrs)
+            // Set Lineage Start to Entry Date and use Identifier as Lineage Start Index for unambiguous ordering
+            .entryDate(entryDate)
+            .lineageStart(entryDate, id)
             .build();
         final StandardRepositoryRecord record = new StandardRepositoryRecord((FlowFileQueue) null);
         record.setWorking(fFile, attrs, false);

nifi-framework-bundle/nifi-framework/nifi-framework-components/src/test/java/org/apache/nifi/controller/repository/StandardProcessSessionTest.java

@@ -45,6 +45,8 @@
 import java.nio.file.Path;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.ArgumentMatchers.anyString;
@@ -191,6 +193,26 @@ void testRecordGaugeSessionCommitted() {
         assertEquals(GAUGE_VALUE, gaugeRecord.value());
     }
 
+    @Test
+    void testCreateLineage() {
+        final long firstFlowFileId = 1;
+        final long secondFlowFileId = 2;
+        when(repositoryContext.getNextFlowFileSequence()).thenReturn(firstFlowFileId, secondFlowFileId);
+
+        final FlowFile firstFlowFile = session.create();
+
+        assertNotNull(firstFlowFile);
+        assertNotEquals(0, firstFlowFile.getLineageStartDate());
+        assertEquals(firstFlowFile.getEntryDate(), firstFlowFile.getLineageStartDate());
+        assertEquals(firstFlowFileId, firstFlowFile.getId());
+        assertEquals(firstFlowFileId, firstFlowFile.getLineageStartIndex());
+
+        final FlowFile secondFlowFile = session.create();
+        assertNotNull(secondFlowFile);
+        assertEquals(secondFlowFileId, secondFlowFile.getId());
+        assertEquals(secondFlowFileId, secondFlowFile.getLineageStartIndex());
+    }
+
     private void assertFlowFileEventMatched(final long bytesRead, final long bytesWritten) throws IOException {
         verify(flowFileEventRepository).updateRepository(flowFileEventCaptor.capture(), anyString());
         final FlowFileEvent flowFileEvent = flowFileEventCaptor.getValue();