@Test @Ignore("Intended only for local performance testing before/after making changes") public void testUpdatePerformance() throws IOException, InterruptedException { final FlowFileQueue queue = new FlowFileQueue() { @Override public String getIdentifier() { return "4444"; } @Override public List<FlowFilePrioritizer> getPriorities() { return null; } @Override public SwapSummary recoverSwappedFlowFiles() { return null; } @Override public void purgeSwapFiles() { } @Override public int getSwapFileCount() { return 0; } @Override public void setPriorities(List<FlowFilePrioritizer> newPriorities) { } @Override public void setBackPressureObjectThreshold(long maxQueueSize) { } @Override public long getBackPressureObjectThreshold() { return 0; } @Override public void setBackPressureDataSizeThreshold(String maxDataSize) { } @Override public String getBackPressureDataSizeThreshold() { return null; } @Override public QueueSize size() { return null; } @Override public boolean isEmpty() { return false; } @Override public boolean isActiveQueueEmpty() { return false; } @Override public QueueSize getUnacknowledgedQueueSize() { return null; } @Override public QueueSize getActiveQueueSize() { return size(); } @Override public QueueSize getSwapQueueSize() { return null; } @Override public void acknowledge(FlowFileRecord flowFile) { } @Override public void acknowledge(Collection<FlowFileRecord> flowFiles) { } @Override public boolean isAllActiveFlowFilesPenalized() { return false; } @Override public boolean isAnyActiveFlowFilePenalized() { return false; } @Override public boolean isFull() { return false; } @Override public void put(FlowFileRecord file) { } @Override public void putAll(Collection<FlowFileRecord> files) { } @Override public FlowFileRecord poll(Set<FlowFileRecord> expiredRecords) { return null; } @Override public List<FlowFileRecord> poll(int maxResults, Set<FlowFileRecord> expiredRecords) { return null; } @Override public long drainQueue(Queue<FlowFileRecord> sourceQueue, List<FlowFileRecord> destination, int maxResults, Set<FlowFileRecord> expiredRecords) { return 0; } @Override public List<FlowFileRecord> poll(FlowFileFilter filter, Set<FlowFileRecord> expiredRecords) { return null; } @Override public String getFlowFileExpiration() { return null; } @Override public int getFlowFileExpiration(TimeUnit timeUnit) { return 0; } @Override public void setFlowFileExpiration(String flowExpirationPeriod) { } @Override public DropFlowFileStatus dropFlowFiles(String requestIdentifier, String requestor) { return null; } @Override public DropFlowFileStatus getDropFlowFileStatus(String requestIdentifier) { return null; } @Override public DropFlowFileStatus cancelDropFlowFileRequest(String requestIdentifier) { return null; } @Override public ListFlowFileStatus listFlowFiles(String requestIdentifier, int maxResults) { return null; } @Override public ListFlowFileStatus getListFlowFileStatus(String requestIdentifier) { return null; } @Override public ListFlowFileStatus cancelListFlowFileRequest(String requestIdentifier) { return null; } @Override public FlowFileRecord getFlowFile(String flowFileUuid) throws IOException { return null; } @Override public void verifyCanList() throws IllegalStateException { } }; final int numPartitions = 16; final int numThreads = 8; final int totalUpdates = 160_000_000; final int batchSize = 10; final Path path = Paths.get("target/minimal-locking-repo"); deleteRecursively(path.toFile()); assertTrue(path.toFile().mkdirs()); final ResourceClaimManager claimManager = new StandardResourceClaimManager(); final RepositoryRecordSerdeFactory serdeFactory = new RepositoryRecordSerdeFactory(claimManager); final WriteAheadRepository<RepositoryRecord> repo = new MinimalLockingWriteAheadLog<>(path, numPartitions, serdeFactory, null); final Collection<RepositoryRecord> initialRecs = repo.recoverRecords(); assertTrue(initialRecs.isEmpty()); final int updateCountPerThread = totalUpdates / numThreads; final Thread[] threads = new Thread[numThreads]; for (int j = 0; j < 2; j++) { for (int i = 0; i < numThreads; i++) { final Thread t = new Thread(new Runnable() { @Override public void run() { final List<RepositoryRecord> records = new ArrayList<>(); final int numBatches = updateCountPerThread / batchSize; final MockFlowFile baseFlowFile = new MockFlowFile(0L); for (int i = 0; i < numBatches; i++) { records.clear(); for (int k = 0; k < batchSize; k++) { final FlowFileRecord flowFile = new MockFlowFile(i % 100_000, baseFlowFile); final String uuid = flowFile.getAttribute("uuid"); final StandardRepositoryRecord record = new StandardRepositoryRecord(null, flowFile); record.setDestination(queue); final Map<String, String> updatedAttrs = Collections.singletonMap("uuid", uuid); record.setWorking(flowFile, updatedAttrs); records.add(record); } try { repo.update(records, false); } catch (IOException e) { e.printStackTrace(); Assert.fail(e.toString()); } } } }); t.setDaemon(true); threads[i] = t; } final long start = System.nanoTime(); for (final Thread t : threads) { t.start(); } for (final Thread t : threads) { t.join(); } final long millis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start); if (j == 0) { System.out.println(millis + " ms to insert " + updateCountPerThread * numThreads + " updates using " + numPartitions + " partitions and " + numThreads + " threads, *as a warmup!*"); } else { System.out.println(millis + " ms to insert " + updateCountPerThread * numThreads + " updates using " + numPartitions + " partitions and " + numThreads + " threads"); } } }
private void deleteRecursively(final File dir) { final File[] children = dir.listFiles(); if (children != null) { for (final File child : children) { final boolean deleted = child.delete(); if (!deleted) { logger.warn("Failed to delete old file {}; this file should be cleaned up manually", child); } } } if (!dir.delete()) { logger.warn("Failed to delete old directory {}; this directory should be cleaned up manually", dir); } }
WriteAheadFlowFileRepository implements FlowFileRepository, SyncListener { private void deleteRecursively(final File dir) { final File[] children = dir.listFiles(); if (children != null) { for (final File child : children) { final boolean deleted = child.delete(); if (!deleted) { logger.warn("Failed to delete old file {}; this file should be cleaned up manually", child); } } } if (!dir.delete()) { logger.warn("Failed to delete old directory {}; this directory should be cleaned up manually", dir); } } }
WriteAheadFlowFileRepository implements FlowFileRepository, SyncListener { private void deleteRecursively(final File dir) { final File[] children = dir.listFiles(); if (children != null) { for (final File child : children) { final boolean deleted = child.delete(); if (!deleted) { logger.warn("Failed to delete old file {}; this file should be cleaned up manually", child); } } } if (!dir.delete()) { logger.warn("Failed to delete old directory {}; this directory should be cleaned up manually", dir); } } WriteAheadFlowFileRepository();  WriteAheadFlowFileRepository(final NiFiProperties nifiProperties); }
WriteAheadFlowFileRepository implements FlowFileRepository, SyncListener { private void deleteRecursively(final File dir) { final File[] children = dir.listFiles(); if (children != null) { for (final File child : children) { final boolean deleted = child.delete(); if (!deleted) { logger.warn("Failed to delete old file {}; this file should be cleaned up manually", child); } } } if (!dir.delete()) { logger.warn("Failed to delete old directory {}; this directory should be cleaned up manually", dir); } } WriteAheadFlowFileRepository();  WriteAheadFlowFileRepository(final NiFiProperties nifiProperties); @Override void initialize(final ResourceClaimManager claimManager); @Override void close(); @Override boolean isVolatile(); @Override long getStorageCapacity(); @Override long getUsableStorageSpace(); @Override String getFileStoreName(); @Override void updateRepository(final Collection<RepositoryRecord> records); @Override void onSync(final int partitionIndex); @Override void onGlobalSync(); @Override void swapFlowFilesOut(final List<FlowFileRecord> swappedOut, final FlowFileQueue queue, final String swapLocation); @Override void swapFlowFilesIn(final String swapLocation, final List<FlowFileRecord> swapRecords, final FlowFileQueue queue); @Override long loadFlowFiles(final QueueProvider queueProvider, final long minimumSequenceNumber); @Override long getNextFlowFileSequence(); @Override long getMaxFlowFileIdentifier(); int checkpoint(); }
WriteAheadFlowFileRepository implements FlowFileRepository, SyncListener { private void deleteRecursively(final File dir) { final File[] children = dir.listFiles(); if (children != null) { for (final File child : children) { final boolean deleted = child.delete(); if (!deleted) { logger.warn("Failed to delete old file {}; this file should be cleaned up manually", child); } } } if (!dir.delete()) { logger.warn("Failed to delete old directory {}; this directory should be cleaned up manually", dir); } } WriteAheadFlowFileRepository();  WriteAheadFlowFileRepository(final NiFiProperties nifiProperties); @Override void initialize(final ResourceClaimManager claimManager); @Override void close(); @Override boolean isVolatile(); @Override long getStorageCapacity(); @Override long getUsableStorageSpace(); @Override String getFileStoreName(); @Override void updateRepository(final Collection<RepositoryRecord> records); @Override void onSync(final int partitionIndex); @Override void onGlobalSync(); @Override void swapFlowFilesOut(final List<FlowFileRecord> swappedOut, final FlowFileQueue queue, final String swapLocation); @Override void swapFlowFilesIn(final String swapLocation, final List<FlowFileRecord> swapRecords, final FlowFileQueue queue); @Override long loadFlowFiles(final QueueProvider queueProvider, final long minimumSequenceNumber); @Override long getNextFlowFileSequence(); @Override long getMaxFlowFileIdentifier(); int checkpoint();  }