<commit_before>
package org.apache.nifi.provenance;

                    ProvenanceEventRepository
                              ^
                              |
                    ProvenanceRepository
                              ^
                              |
                    PersistentProvenanceRepository


Startup
====================
1. nifi.properties - typical configuration is
    nifi.provenance.repository.implementation=org.apache.nifi.provenance.PersistentProvenanceRepository

2. NifiProperties.java - pulls it in
    public static final PROVENANCE_REPO_IMPLEMENTATION_CLASS = "nifi.provenance.repository.implementation";

3. FlowController.java
    public static final String PROVENANCE_REPO_IMPLEMENTATION_CLASS = "nifi.provenance.repository.implementation";
    public static final String DEFAULT_PROVENANCE_REPO_IMPLEMENTATION = "org.apache.nifi.provenance.VolatileProvenanceRepository";

    private FlowController(...) {
    ...
        try {
            this.provenanceRepository = createProvenanceRepository(nifiProperties);
            this.provenanceRepository.initialize(createEventReporter(bulletinRepository), authorizer, this);
        } catch (final Exception e) {
            throw new RuntimeException("Unable to create Provenance Repository", e);
        }
    ...
    }

    private ProvenanceRepository createProvenanceRepository(final NiFiProperties properties) throws InstantiationException, IllegalAccessException, ClassNotFoundException {
  **    final String implementationClassName = properties.getProperty(NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS, DEFAULT_PROVENANCE_REPO_IMPLEMENTATION);
        if (implementationClassName == null) {
            throw new RuntimeException("Cannot create Provenance Repository because the NiFi Properties is missing the following property: "
                    + NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS);
        }
    ...
        try {
  **        return NarThreadContextClassLoader.createInstance(implementationClassName, ProvenanceRepository.class, properties);
        } catch (final Exception e) {
            throw new RuntimeException(e);
        }
    }

4. PersistentProvenanceRepository.java
    public PersistentProvenanceRepository(final NiFiProperties nifiProperties) throws IOException {
  **    this(createRepositoryConfiguration(nifiProperties), 10000);
    }

  **public PersistentProvenanceRepository(final RepositoryConfiguration configuration, final int rolloverCheckMillis) throws IOException {
        if (configuration.getStorageDirectories().isEmpty()) {
            throw new IllegalArgumentException("Must specify at least one storage directory");
        }

        this.configuration = configuration;
        this.maxAttributeChars = configuration.getMaxAttributeChars();

        for (final File file : configuration.getStorageDirectories()) {
            final Path storageDirectory = file.toPath();
            final Path journalDirectory = storageDirectory.resolve("journals");

            if (!Files.exists(journalDirectory)) {
                Files.createDirectories(journalDirectory);
            } else if (!Files.isDirectory(journalDirectory)) {
                throw new IllegalArgumentException("Storage Location " + journalDirectory + " is not a directory");
            }
        }

        this.maxPartitionMillis = configuration.getMaxEventFileLife(TimeUnit.MILLISECONDS);
        this.maxPartitionBytes = configuration.getMaxEventFileCapacity();
        this.indexConfig = new IndexConfiguration(configuration);
        this.indexManager = new SimpleIndexManager();
        this.alwaysSync = configuration.isAlwaysSync();
        this.rolloverCheckMillis = rolloverCheckMillis;

        scheduledExecService = Executors.newScheduledThreadPool(3, new NamedThreadFactory("Provenance Maintenance Thread"));
        queryExecService = Executors.newFixedThreadPool(configuration.getQueryThreadPoolSize(), new NamedThreadFactory("Provenance Query Thread"));

        // The number of rollover threads is a little bit arbitrary but comes from the idea that multiple storage directories generally
        // live on separate physical partitions. As a result, we want to use at least one thread per partition in order to utilize the
        // disks efficiently. However, the rollover actions can be somewhat CPU intensive, so we double the number of threads in order
        // to account for that.
        final int numRolloverThreads = configuration.getStorageDirectories().size() * 2;
        rolloverExecutor = Executors.newScheduledThreadPool(numRolloverThreads, new NamedThreadFactory("Provenance Repository Rollover Thread"));
    }

    @Override
    public void initialize(final EventReporter eventReporter, final Authorizer authorizer, final ProvenanceAuthorizableFactory resourceFactory) throws IOException {
        writeLock.lock();
        try {
            if (initialized.getAndSet(true)) {
                return;
            }
     ...
  **        recover();
            if (configuration.isAllowRollover()) {
                writers = createWriters(configuration, idGenerator.get());
            }

            if (configuration.isAllowRollover()) {
                scheduledExecService.scheduleWithFixedDelay(new Runnable() {
                    @Override public void run() {
                        // Check if we need to roll over
                        if (needToRollover()) {
                            // it appears that we do need to roll over. Obtain write lock so that we can do so, and then
                            // confirm that we still need to.
                            writeLock.lock();
                            try {
                                logger.debug("Obtained write lock to perform periodic rollover");
                                if (needToRollover()) {
                                    try {
                                        rollover(false);
                                    } catch (final Exception e) {
                                        logger.error("Failed to roll over Provenance Event Log due to {}", e.toString());
                                        logger.error("", e);
                                        eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to roll over Provenance Event Log due to " + e.toString());
                                    }
                                }
                            } finally {
                                writeLock.unlock();
                            }
                        }
                    }
                }, rolloverCheckMillis, rolloverCheckMillis, TimeUnit.MILLISECONDS);

                expirationActions.add(new UpdateMinimumEventId(indexConfig));
                expirationActions.add(new FileRemovalAction());

                scheduledExecService.scheduleWithFixedDelay(new RemoveExpiredQueryResults(), 30L, 3L, TimeUnit.SECONDS);
                scheduledExecService.scheduleWithFixedDelay(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            purgeOldEvents();
                        } catch (final Exception e) {
                            logger.error("Failed to purge old events from Provenance Repo due to {}", e.toString());
                            if (logger.isDebugEnabled()) {
                                logger.error("", e);
                            }
                            eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to purge old events from Provenance Repo due to " + e.toString());
                        }
                    }
                }, PURGE_EVENT_MILLISECONDS, PURGE_EVENT_MILLISECONDS, TimeUnit.MILLISECONDS);
            }

            firstEventTimestamp = determineFirstEventTimestamp();
        } finally {
            writeLock.unlock();
        }
    }



public interface ProvenanceEventRepository {
  void                        close                        () throws IOException;
  ProvenanceEventBuilder      eventBuilder                 ();
  ProvenanceEventRecord       getEvent                     (long id) throws IOException;
  List<ProvenanceEventRecord> getEvents                    (long firstRecordId, final int maxRecords) throws IOException;
  Long                        getMaxEventId                ();
  void                        registerEvent                (ProvenanceEventRecord event);
  void                        registerEvents               (Iterable<ProvenanceEventRecord> events);
}

public interface ProvenanceRepository extends ProvenanceEventRepository {
  ProvenanceEventRepository   getProvenanceEventRepository ();
  void                        initialize                   (EventReporter eventReporter, Authorizer authorizer, ProvenanceAuthorizableFactory resourceFactory) throws IOException;

  ProvenanceEventRecord       getEvent                     (long id, NiFiUser user) throws IOException;
  List<ProvenanceEventRecord> getEvents                    (long firstRecordId, final int maxRecords, NiFiUser user) throws IOException;
  List<SearchableField>       getSearchableAttributes      ();
  List<SearchableField>       getSearchableFields          ();
  ComputeLineageSubmission    retrieveLineageSubmission    (String lineageIdentifier, NiFiUser user);
  QuerySubmission             retrieveQuerySubmission      (String queryIdentifier, NiFiUser user);
  ComputeLineageSubmission    submitLineageComputation     (String flowFileUuid, NiFiUser user);
  ComputeLineageSubmission    submitLineageComputation     (long eventId, NiFiUser user);
  ComputeLineageSubmission    submitExpandParents          (long eventId, NiFiUser user);
  ComputeLineageSubmission    submitExpandChildren         (long eventId, NiFiUser user);
  QuerySubmission             submitQuery                  (Query query, NiFiUser user);
}<commit_msg>Provenance edits.<commit_after>
package org.apache.nifi.provenance;

PersistentProvenanceRepository
    |
    | implements
    v
    ProvenanceRepository
        |
                    ProvenanceEventRepository



Startup
====================
1. nifi.properties - typical configuration is
    nifi.provenance.repository.implementation=org.apache.nifi.provenance.PersistentProvenanceRepository

2. NifiProperties.java - pulls it in
    public static final PROVENANCE_REPO_IMPLEMENTATION_CLASS = "nifi.provenance.repository.implementation";

3. FlowController.java
    - FlowController() constructor calls createProvenanceRepository(nifiProperties)
    - createProvenanceRepository() calls NarThreadContextClassLoader.createInstance(implementationClassName, ProvenanceRepository.class, properties)

    public static final String PROVENANCE_REPO_IMPLEMENTATION_CLASS = "nifi.provenance.repository.implementation";
    public static final String DEFAULT_PROVENANCE_REPO_IMPLEMENTATION = "org.apache.nifi.provenance.VolatileProvenanceRepository";

    private FlowController(...) {
    ...
        try {
**          this.provenanceRepository = createProvenanceRepository(nifiProperties);
            this.provenanceRepository.initialize(createEventReporter(bulletinRepository),
                                                 authorizer, this);
        } catch (final Exception e) {
            throw new RuntimeException("Unable to create Provenance Repository", e);
        }
    ...
    }

    private ProvenanceRepository createProvenanceRepository(final NiFiProperties properties) throws InstantiationException, IllegalAccessException, ClassNotFoundException {
**      final String implementationClassName = properties.getProperty(NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS, DEFAULT_PROVENANCE_REPO_IMPLEMENTATION);
        if (implementationClassName == null) {
            throw new RuntimeException("Cannot create Provenance Repository because the NiFi Properties is missing the following property: "
                    + NiFiProperties.PROVENANCE_REPO_IMPLEMENTATION_CLASS);
        }
    ...
        try {
**          return NarThreadContextClassLoader.createInstance(implementationClassName, ProvenanceRepository.class, properties);
        } catch (final Exception e) {
            throw new RuntimeException(e);
        }
    }

4. PersistentProvenanceRepository.java
    public PersistentProvenanceRepository(final NiFiProperties nifiProperties) throws IOException {
**      this(createRepositoryConfiguration(nifiProperties), 10000);
    }

**  public PersistentProvenanceRepository(final RepositoryConfiguration configuration, final int rolloverCheckMillis) throws IOException {
        if (configuration.getStorageDirectories().isEmpty()) {
            throw new IllegalArgumentException("Must specify at least one storage directory");
        }

        this.configuration = configuration;
        this.maxAttributeChars = configuration.getMaxAttributeChars();

        for (final File file : configuration.getStorageDirectories()) {
            final Path storageDirectory = file.toPath();
            final Path journalDirectory = storageDirectory.resolve("journals");

            if (!Files.exists(journalDirectory)) {
                Files.createDirectories(journalDirectory);
            } else if (!Files.isDirectory(journalDirectory)) {
                throw new IllegalArgumentException("Storage Location " + journalDirectory + " is not a directory");
            }
        }

        this.maxPartitionMillis = configuration.getMaxEventFileLife(TimeUnit.MILLISECONDS);
        this.maxPartitionBytes = configuration.getMaxEventFileCapacity();
        this.indexConfig = new IndexConfiguration(configuration);
        this.indexManager = new SimpleIndexManager();
        this.alwaysSync = configuration.isAlwaysSync();
        this.rolloverCheckMillis = rolloverCheckMillis;

        scheduledExecService = Executors.newScheduledThreadPool(3, new NamedThreadFactory("Provenance Maintenance Thread"));
        queryExecService = Executors.newFixedThreadPool(configuration.getQueryThreadPoolSize(), new NamedThreadFactory("Provenance Query Thread"));

        // The number of rollover threads is a little bit arbitrary but comes from the idea that multiple storage directories generally
        // live on separate physical partitions. As a result, we want to use at least one thread per partition in order to utilize the
        // disks efficiently. However, the rollover actions can be somewhat CPU intensive, so we double the number of threads in order
        // to account for that.
        final int numRolloverThreads = configuration.getStorageDirectories().size() * 2;
        rolloverExecutor = Executors.newScheduledThreadPool(numRolloverThreads, new NamedThreadFactory("Provenance Repository Rollover Thread"));
    }

    @Override
    public void initialize(final EventReporter eventReporter, final Authorizer authorizer, final ProvenanceAuthorizableFactory resourceFactory) throws IOException {
        writeLock.lock();
        try {
            if (initialized.getAndSet(true)) {
                return;
            }
     ...
  **        recover();
            if (configuration.isAllowRollover()) {
                writers = createWriters(configuration, idGenerator.get());
            }

            if (configuration.isAllowRollover()) {
                scheduledExecService.scheduleWithFixedDelay(new Runnable() {
                    @Override public void run() {
                        // Check if we need to roll over
                        if (needToRollover()) {
                            // it appears that we do need to roll over. Obtain write lock so that we can do so, and then
                            // confirm that we still need to.
                            writeLock.lock();
                            try {
                                logger.debug("Obtained write lock to perform periodic rollover");
                                if (needToRollover()) {
                                    try {
                                        rollover(false);
                                    } catch (final Exception e) {
                                        logger.error("Failed to roll over Provenance Event Log due to {}", e.toString());
                                        logger.error("", e);
                                        eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to roll over Provenance Event Log due to " + e.toString());
                                    }
                                }
                            } finally {
                                writeLock.unlock();
                            }
                        }
                    }
                }, rolloverCheckMillis, rolloverCheckMillis, TimeUnit.MILLISECONDS);

                expirationActions.add(new UpdateMinimumEventId(indexConfig));
                expirationActions.add(new FileRemovalAction());

                scheduledExecService.scheduleWithFixedDelay(new RemoveExpiredQueryResults(), 30L, 3L, TimeUnit.SECONDS);
                scheduledExecService.scheduleWithFixedDelay(new Runnable() {
                    @Override
                    public void run() {
                        try {
                            purgeOldEvents();
                        } catch (final Exception e) {
                            logger.error("Failed to purge old events from Provenance Repo due to {}", e.toString());
                            if (logger.isDebugEnabled()) {
                                logger.error("", e);
                            }
                            eventReporter.reportEvent(Severity.ERROR, EVENT_CATEGORY, "Failed to purge old events from Provenance Repo due to " + e.toString());
                        }
                    }
                }, PURGE_EVENT_MILLISECONDS, PURGE_EVENT_MILLISECONDS, TimeUnit.MILLISECONDS);
            }

            firstEventTimestamp = determineFirstEventTimestamp();
        } finally {
            writeLock.unlock();
        }
    }

===========================================================================

public interface ProvenanceEventRepository {
  void                        close                        () throws IOException;
  ProvenanceEventBuilder      eventBuilder                 ();
  ProvenanceEventRecord       getEvent                     (long id) throws IOException;
  List<ProvenanceEventRecord> getEvents                    (long firstRecordId, final int maxRecords) throws IOException;
  Long                        getMaxEventId                ();
  void                        registerEvent                (ProvenanceEventRecord event);
  void                        registerEvents               (Iterable<ProvenanceEventRecord> events);
}

public interface ProvenanceRepository extends ProvenanceEventRepository {
  ProvenanceEventRepository   getProvenanceEventRepository ();
  void                        initialize                   (EventReporter eventReporter, Authorizer authorizer, ProvenanceAuthorizableFactory resourceFactory) throws IOException;

  ProvenanceEventRecord       getEvent                     (long id, NiFiUser user) throws IOException;
  List<ProvenanceEventRecord> getEvents                    (long firstRecordId, final int maxRecords, NiFiUser user) throws IOException;
  List<SearchableField>       getSearchableAttributes      ();
  List<SearchableField>       getSearchableFields          ();
  ComputeLineageSubmission    retrieveLineageSubmission    (String lineageIdentifier, NiFiUser user);
  QuerySubmission             retrieveQuerySubmission      (String queryIdentifier, NiFiUser user);
  ComputeLineageSubmission    submitLineageComputation     (String flowFileUuid, NiFiUser user);
  ComputeLineageSubmission    submitLineageComputation     (long eventId, NiFiUser user);
  ComputeLineageSubmission    submitExpandParents          (long eventId, NiFiUser user);
  ComputeLineageSubmission    submitExpandChildren         (long eventId, NiFiUser user);
  QuerySubmission             submitQuery                  (Query query, NiFiUser user);
}
{'fraction_non_alphanumeric': 0.07107478949975235, 'fraction_numerical': 0.0013868251609707777, 'mean_word_length': 2.1342750698540827, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 18, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2930790', 'n_tokens_mistral': 4618, 'n_tokens_neox': 4447, 'n_words': 1122}