@Test public void testTruncateAttributes() throws IOException, InterruptedException { final RepositoryConfiguration config = createConfiguration(); config.setMaxAttributeChars(50); config.setMaxEventFileLife(3, TimeUnit.SECONDS); repo = new PersistentProvenanceRepository(config, DEFAULT_ROLLOVER_MILLIS); repo.initialize(getEventReporter(), null, null, IdentifierLookup.EMPTY); final String maxLengthChars = "12345678901234567890123456789012345678901234567890"; final Map<String, String> attributes = new HashMap<>(); attributes.put("75chars", "123456789012345678901234567890123456789012345678901234567890123456789012345"); attributes.put("51chars", "123456789012345678901234567890123456789012345678901"); attributes.put("50chars", "12345678901234567890123456789012345678901234567890"); attributes.put("49chars", "1234567890123456789012345678901234567890123456789"); attributes.put("nullChar", null); final ProvenanceEventBuilder builder = new StandardProvenanceEventRecord.Builder(); builder.setEventTime(System.currentTimeMillis()); builder.setEventType(ProvenanceEventType.RECEIVE); builder.setTransitUri("nifi: attributes.put("uuid", "12345678-0000-0000-0000-012345678912"); builder.fromFlowFile(createFlowFile(3L, 3000L, attributes)); builder.setComponentId("1234"); builder.setComponentType("dummy processor"); final ProvenanceEventRecord record = builder.build(); repo.registerEvent(record); repo.waitForRollover(); final ProvenanceEventRecord retrieved = repo.getEvent(0L, null); assertNotNull(retrieved); assertEquals("12345678-0000-0000-0000-012345678912", retrieved.getAttributes().get("uuid")); assertEquals(maxLengthChars, retrieved.getAttributes().get("75chars")); assertEquals(maxLengthChars, retrieved.getAttributes().get("51chars")); assertEquals(maxLengthChars, retrieved.getAttributes().get("50chars")); assertEquals(maxLengthChars.substring(0, 49), retrieved.getAttributes().get("49chars")); }
private StandardProvenanceEventRecord truncateAttributes(final StandardProvenanceEventRecord original) { boolean requireTruncation = false; for (final String updatedAttr : original.getUpdatedAttributes().values()) { if (updatedAttr != null && updatedAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } if (!requireTruncation) { for (final String previousAttr : original.getPreviousAttributes().values()) { if (previousAttr != null && previousAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } } if (!requireTruncation) { return original; } final StandardProvenanceEventRecord.Builder builder = new StandardProvenanceEventRecord.Builder().fromEvent(original); builder.setAttributes(truncateAttributes(original.getPreviousAttributes()), truncateAttributes(original.getUpdatedAttributes())); final StandardProvenanceEventRecord truncated = builder.build(); truncated.setEventId(original.getEventId()); return truncated; }
PersistentProvenanceRepository implements ProvenanceRepository { private StandardProvenanceEventRecord truncateAttributes(final StandardProvenanceEventRecord original) { boolean requireTruncation = false; for (final String updatedAttr : original.getUpdatedAttributes().values()) { if (updatedAttr != null && updatedAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } if (!requireTruncation) { for (final String previousAttr : original.getPreviousAttributes().values()) { if (previousAttr != null && previousAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } } if (!requireTruncation) { return original; } final StandardProvenanceEventRecord.Builder builder = new StandardProvenanceEventRecord.Builder().fromEvent(original); builder.setAttributes(truncateAttributes(original.getPreviousAttributes()), truncateAttributes(original.getUpdatedAttributes())); final StandardProvenanceEventRecord truncated = builder.build(); truncated.setEventId(original.getEventId()); return truncated; } }
PersistentProvenanceRepository implements ProvenanceRepository { private StandardProvenanceEventRecord truncateAttributes(final StandardProvenanceEventRecord original) { boolean requireTruncation = false; for (final String updatedAttr : original.getUpdatedAttributes().values()) { if (updatedAttr != null && updatedAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } if (!requireTruncation) { for (final String previousAttr : original.getPreviousAttributes().values()) { if (previousAttr != null && previousAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } } if (!requireTruncation) { return original; } final StandardProvenanceEventRecord.Builder builder = new StandardProvenanceEventRecord.Builder().fromEvent(original); builder.setAttributes(truncateAttributes(original.getPreviousAttributes()), truncateAttributes(original.getUpdatedAttributes())); final StandardProvenanceEventRecord truncated = builder.build(); truncated.setEventId(original.getEventId()); return truncated; } PersistentProvenanceRepository();  PersistentProvenanceRepository(final NiFiProperties nifiProperties);  PersistentProvenanceRepository(final RepositoryConfiguration configuration, final int rolloverCheckMillis); }
PersistentProvenanceRepository implements ProvenanceRepository { private StandardProvenanceEventRecord truncateAttributes(final StandardProvenanceEventRecord original) { boolean requireTruncation = false; for (final String updatedAttr : original.getUpdatedAttributes().values()) { if (updatedAttr != null && updatedAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } if (!requireTruncation) { for (final String previousAttr : original.getPreviousAttributes().values()) { if (previousAttr != null && previousAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } } if (!requireTruncation) { return original; } final StandardProvenanceEventRecord.Builder builder = new StandardProvenanceEventRecord.Builder().fromEvent(original); builder.setAttributes(truncateAttributes(original.getPreviousAttributes()), truncateAttributes(original.getUpdatedAttributes())); final StandardProvenanceEventRecord truncated = builder.build(); truncated.setEventId(original.getEventId()); return truncated; } PersistentProvenanceRepository();  PersistentProvenanceRepository(final NiFiProperties nifiProperties);  PersistentProvenanceRepository(final RepositoryConfiguration configuration, final int rolloverCheckMillis); @Override void initialize(final EventReporter eventReporter, final Authorizer authorizer, final ProvenanceAuthorizableFactory resourceFactory,
        final IdentifierLookup idLookup); int getMaxAttributeCharacters(); @Override StandardProvenanceEventRecord.Builder eventBuilder(); @Override void registerEvent(final ProvenanceEventRecord event); @Override void registerEvents(final Iterable<ProvenanceEventRecord> events); boolean isAuthorized(final ProvenanceEventRecord event, final NiFiUser user); void authorize(final ProvenanceEventRecord event, final NiFiUser user); List<ProvenanceEventRecord> filterUnauthorizedEvents(final List<ProvenanceEventRecord> events, final NiFiUser user); Set<ProvenanceEventRecord> replaceUnauthorizedWithPlaceholders(final Set<ProvenanceEventRecord> events, final NiFiUser user); @Override List<ProvenanceEventRecord> getEvents(final long firstRecordId, final int maxRecords); @Override List<ProvenanceEventRecord> getEvents(final long firstRecordId, final int maxRecords, final NiFiUser user); RepositoryConfiguration getConfiguration(); @Override synchronized void close(); boolean isShutdownComplete(); long getSize(final List<File> logFiles, final long timeCutoff); void waitForRollover(); @Override List<SearchableField> getSearchableFields(); @Override List<SearchableField> getSearchableAttributes(); @Override QuerySubmission submitQuery(final Query query, final NiFiUser user); @Override ComputeLineageSubmission submitLineageComputation(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission submitLineageComputation(final String flowFileUuid, final NiFiUser user); @Override AsyncLineageSubmission submitExpandChildren(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission submitExpandParents(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission retrieveLineageSubmission(final String lineageIdentifier, final NiFiUser user); @Override QuerySubmission retrieveQuerySubmission(final String queryIdentifier, final NiFiUser user); @Override ProvenanceEventRecord getEvent(final long id); @Override ProvenanceEventRecord getEvent(final long id, final NiFiUser user); @Override ProvenanceEventRepository getProvenanceEventRepository(); Collection<Path> getAllLogFiles(); @Override Long getMaxEventId(); }
PersistentProvenanceRepository implements ProvenanceRepository { private StandardProvenanceEventRecord truncateAttributes(final StandardProvenanceEventRecord original) { boolean requireTruncation = false; for (final String updatedAttr : original.getUpdatedAttributes().values()) { if (updatedAttr != null && updatedAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } if (!requireTruncation) { for (final String previousAttr : original.getPreviousAttributes().values()) { if (previousAttr != null && previousAttr.length() > maxAttributeChars) { requireTruncation = true; break; } } } if (!requireTruncation) { return original; } final StandardProvenanceEventRecord.Builder builder = new StandardProvenanceEventRecord.Builder().fromEvent(original); builder.setAttributes(truncateAttributes(original.getPreviousAttributes()), truncateAttributes(original.getUpdatedAttributes())); final StandardProvenanceEventRecord truncated = builder.build(); truncated.setEventId(original.getEventId()); return truncated; } PersistentProvenanceRepository();  PersistentProvenanceRepository(final NiFiProperties nifiProperties);  PersistentProvenanceRepository(final RepositoryConfiguration configuration, final int rolloverCheckMillis); @Override void initialize(final EventReporter eventReporter, final Authorizer authorizer, final ProvenanceAuthorizableFactory resourceFactory,
        final IdentifierLookup idLookup); int getMaxAttributeCharacters(); @Override StandardProvenanceEventRecord.Builder eventBuilder(); @Override void registerEvent(final ProvenanceEventRecord event); @Override void registerEvents(final Iterable<ProvenanceEventRecord> events); boolean isAuthorized(final ProvenanceEventRecord event, final NiFiUser user); void authorize(final ProvenanceEventRecord event, final NiFiUser user); List<ProvenanceEventRecord> filterUnauthorizedEvents(final List<ProvenanceEventRecord> events, final NiFiUser user); Set<ProvenanceEventRecord> replaceUnauthorizedWithPlaceholders(final Set<ProvenanceEventRecord> events, final NiFiUser user); @Override List<ProvenanceEventRecord> getEvents(final long firstRecordId, final int maxRecords); @Override List<ProvenanceEventRecord> getEvents(final long firstRecordId, final int maxRecords, final NiFiUser user); RepositoryConfiguration getConfiguration(); @Override synchronized void close(); boolean isShutdownComplete(); long getSize(final List<File> logFiles, final long timeCutoff); void waitForRollover(); @Override List<SearchableField> getSearchableFields(); @Override List<SearchableField> getSearchableAttributes(); @Override QuerySubmission submitQuery(final Query query, final NiFiUser user); @Override ComputeLineageSubmission submitLineageComputation(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission submitLineageComputation(final String flowFileUuid, final NiFiUser user); @Override AsyncLineageSubmission submitExpandChildren(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission submitExpandParents(final long eventId, final NiFiUser user); @Override AsyncLineageSubmission retrieveLineageSubmission(final String lineageIdentifier, final NiFiUser user); @Override QuerySubmission retrieveQuerySubmission(final String queryIdentifier, final NiFiUser user); @Override ProvenanceEventRecord getEvent(final long id); @Override ProvenanceEventRecord getEvent(final long id, final NiFiUser user); @Override ProvenanceEventRepository getProvenanceEventRepository(); Collection<Path> getAllLogFiles(); @Override Long getMaxEventId(); static final String EVENT_CATEGORY; static final Pattern NUMBER_PATTERN; static final Pattern INDEX_PATTERN; static final Pattern LOG_FILENAME_PATTERN; static final int MAX_UNDELETED_QUERY_RESULTS; static final int MAX_INDEXING_FAILURE_COUNT; static final int MAX_JOURNAL_ROLLOVER_RETRIES; }