NIFI-13638: Limiting the maximum number of provenance events to return when fetching the latest events. (#9156)

minifi/minifi-nar-bundles/minifi-provenance-repository-bundle/minifi-provenance-repositories/src/main/java/org/apache/nifi/provenance/NoOpProvenanceRepository.java

@@ -103,7 +103,7 @@ public class NoOpProvenanceRepository implements ProvenanceRepository {
   }
 
   @Override
-  public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId) {
+  public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId, final int eventLimit) {
     return List.of();
   }
 

nifi-extension-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/WriteAheadProvenanceRepository.java

@@ -257,8 +257,8 @@ public class WriteAheadProvenanceRepository implements ProvenanceRepository {
     }
 
     @Override
-    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId) throws IOException {
+    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId, final int eventLimit) throws IOException {
-        return eventIndex.getLatestCachedEvents(componentId);
+        return eventIndex.getLatestCachedEvents(componentId, eventLimit);
     }
 
     @Override

nifi-extension-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/EventIndex.java

@@ -85,11 +85,12 @@ public interface EventIndex extends Closeable {
     /**
      * Retrieves the list of Provenance Events that are cached for the most recent invocation of the given component
      * @param componentId the ID of the component
+     * @param eventLimit the maximum number of events to return
      *
      * @return an Optional containing the event, or an empty optional if no events are available or none of the available events are accessible by the given user
      * @throws IOException if unable to read from the repository
      */
-    List<ProvenanceEventRecord> getLatestCachedEvents(String componentId) throws IOException;
+    List<ProvenanceEventRecord> getLatestCachedEvents(String componentId, int eventLimit) throws IOException;
 
     /**
      * Asynchronously computes the lineage for the FlowFile that is identified by the Provenance Event with the given ID.

nifi-extension-bundles/nifi-provenance-repository-bundle/nifi-persistent-provenance-repository/src/main/java/org/apache/nifi/provenance/index/lucene/LuceneEventIndex.java

@@ -643,15 +643,17 @@ public class LuceneEventIndex implements EventIndex {
     }
 
     @Override
-    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId) throws IOException {
+    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId, final int eventLimit) throws IOException {
         final List<Long> eventIds = latestEventsPerProcessorQuery.getLatestEventIds(componentId);
         if (eventIds.isEmpty()) {
             logger.info("There are no recent Provenance Events cached for Component with ID {}", componentId);
             return List.of();
         }
 
-        final List<ProvenanceEventRecord> latestEvents = new ArrayList<>(eventIds.size());
+        final List<Long> filtered = eventIds.subList(0, Math.min(eventIds.size(), eventLimit));
-        for (final Long eventId : eventIds) {
+
+        final List<ProvenanceEventRecord> latestEvents = new ArrayList<>(filtered.size());
+        for (final Long eventId : filtered) {
             final Optional<ProvenanceEventRecord> latestEvent = eventStore.getEvent(eventId);
             if (latestEvent.isPresent()) {
                 latestEvents.add(latestEvent.get());

nifi-extension-bundles/nifi-provenance-repository-bundle/nifi-volatile-provenance-repository/src/main/java/org/apache/nifi/provenance/VolatileProvenanceRepository.java

@@ -491,7 +491,7 @@ public class VolatileProvenanceRepository implements ProvenanceRepository {
     }
 
     @Override
-    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId) {
+    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId, final int eventLimit) {
         final List<ProvenanceEventRecord> matches = ringBuffer.getSelectedElements(
             event -> componentId.equals(event.getComponentId()), 1);
 

nifi-framework-api/src/main/java/org/apache/nifi/provenance/ProvenanceRepository.java

@@ -96,10 +96,11 @@ public interface ProvenanceRepository extends ProvenanceEventRepository {
     /**
      * Retrieves the Provenance Events that are cached for the most recent invocation of the given component.
      * @param componentId the ID of the component
+     * @param eventLimit the maximum number of events to return
      * @return the list of events that are cached for the given component
      * @throws IOException if unable to read from the repository
      */
-    List<ProvenanceEventRecord> getLatestCachedEvents(String componentId) throws IOException;
+    List<ProvenanceEventRecord> getLatestCachedEvents(String componentId, int eventLimit) throws IOException;
 
     /**
      * @param queryIdentifier of the query

nifi-framework-bundle/nifi-framework/nifi-framework-core/src/test/java/org/apache/nifi/provenance/MockProvenanceRepository.java

@@ -91,7 +91,7 @@ public class MockProvenanceRepository implements ProvenanceRepository {
     }
 
     @Override
-    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId) {
+    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId, final int eventLimit) {
         return List.of();
     }
 

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/NiFiServiceFacade.java

@@ -326,9 +326,10 @@ public interface NiFiServiceFacade {
     /**
      * Gets the latest provenance events for the specified component.
      * @param componentId the ID of the components to retrieve the latest events for
+     * @param eventLimit the maximum number of events to return
      * @return the latest provenance events
      */
-    LatestProvenanceEventsEntity getLatestProvenanceEvents(String componentId);
+    LatestProvenanceEventsEntity getLatestProvenanceEvents(String componentId, int eventLimit);
 
     /**
      * Gets the configuration for this controller.

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/StandardNiFiServiceFacade.java

@@ -3663,8 +3663,8 @@ public class StandardNiFiServiceFacade implements NiFiServiceFacade {
     }
 
     @Override
-    public LatestProvenanceEventsEntity getLatestProvenanceEvents(final String componentId) {
+    public LatestProvenanceEventsEntity getLatestProvenanceEvents(final String componentId, final int eventLimit) {
-        final LatestProvenanceEventsDTO dto = controllerFacade.getLatestProvenanceEvents(componentId);
+        final LatestProvenanceEventsDTO dto = controllerFacade.getLatestProvenanceEvents(componentId, eventLimit);
 
         final LatestProvenanceEventsEntity entity = new LatestProvenanceEventsEntity();
         entity.setLatestProvenanceEvents(dto);

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProvenanceEventResource.java

@@ -26,6 +26,7 @@ import io.swagger.v3.oas.annotations.security.SecurityRequirement;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import jakarta.servlet.http.HttpServletRequest;
 import jakarta.ws.rs.Consumes;
+import jakarta.ws.rs.DefaultValue;
 import jakarta.ws.rs.GET;
 import jakarta.ws.rs.HttpMethod;
 import jakarta.ws.rs.POST;
@@ -526,14 +527,19 @@ public class ProvenanceEventResource extends ApplicationResource {
             description = "The ID of the component to retrieve the latest Provenance Events for.",
             required = true
         )
-        @PathParam("componentId") final String componentId
+        @PathParam("componentId") final String componentId,
+        @Parameter(
+                description = "The number of events to limit the response to. Defaults to 10."
+        )
+        @DefaultValue("10")
+        @QueryParam("limit") int limit
     ) {
         if (isReplicateRequest()) {
             return replicate(HttpMethod.GET);
         }
 
         // get the latest provenance events
-        final LatestProvenanceEventsEntity entity = serviceFacade.getLatestProvenanceEvents(componentId);
+        final LatestProvenanceEventsEntity entity = serviceFacade.getLatestProvenanceEvents(componentId, limit);
 
         // generate the response
         return generateOkResponse(entity).build();

nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/controller/ControllerFacade.java

@@ -155,6 +155,8 @@ public class ControllerFacade implements Authorizable {
 
     private static final Logger logger = LoggerFactory.getLogger(ControllerFacade.class);
 
+    private static final int MAX_REPLAY_EVENT_COUNT = 10;
+
     // nifi components
     private FlowController flowController;
     private FlowService flowService;
@@ -1415,7 +1417,7 @@ public class ControllerFacade implements Authorizable {
             }
 
             // lookup the original event
-            final List<ProvenanceEventRecord> latestEvents = flowController.getProvenanceRepository().getLatestCachedEvents(componentId);
+            final List<ProvenanceEventRecord> latestEvents = flowController.getProvenanceRepository().getLatestCachedEvents(componentId, MAX_REPLAY_EVENT_COUNT);
             if (latestEvents.isEmpty()) {
                 return null;
             }
@@ -1537,7 +1539,7 @@ public class ControllerFacade implements Authorizable {
         }
     }
 
-    public LatestProvenanceEventsDTO getLatestProvenanceEvents(final String componentId) {
+    public LatestProvenanceEventsDTO getLatestProvenanceEvents(final String componentId, final int eventLimit) {
         final Authorizable authorizable = flowController.getProvenanceAuthorizableFactory().createProvenanceDataAuthorizable(componentId);
         final Authorizer authorizer = flowController.getAuthorizer();
         if (!authorizable.isAuthorized(authorizer, RequestAction.READ, NiFiUserUtils.getNiFiUser())) {
@@ -1545,7 +1547,7 @@ public class ControllerFacade implements Authorizable {
         }
 
         try {
-            final List<ProvenanceEventRecord> events = flowController.getProvenanceRepository().getLatestCachedEvents(componentId);
+            final List<ProvenanceEventRecord> events = flowController.getProvenanceRepository().getLatestCachedEvents(componentId, eventLimit);
             final List<ProvenanceEventDTO> eventDtos = new ArrayList<>();
             for (final ProvenanceEventRecord event : events) {
                 eventDtos.add(createProvenanceEventDto(event, false));

nifi-stateless/nifi-stateless-bundle/nifi-stateless-engine/src/main/java/org/apache/nifi/stateless/repository/StatelessProvenanceRepository.java

@@ -153,7 +153,7 @@ public class StatelessProvenanceRepository implements ProvenanceRepository {
     }
 
     @Override
-    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId) {
+    public List<ProvenanceEventRecord> getLatestCachedEvents(final String componentId, final int eventLimit) {
         return List.of();
     }