@Test public void testPendingEncryptJobs() throws InterruptedException, GeneralSecurityException { nettyByteBufLeakHelper.setDisabled(true); int testDataCount = 10; int closeOnCount = 4; CountDownLatch encryptCallBackCount = new CountDownLatch(testDataCount); List<EncryptJob> encryptJobs = new ArrayList<>(); SecretKeySpec perBlobKey = kms.getRandomKey(); for (int i = 0; i < testDataCount; i++) { TestBlobData testData = getRandomBlob(referenceClusterMap); if (i < closeOnCount) { cryptoJobHandler.submitJob( new EncryptJob(testData.blobId.getAccountId(), testData.blobId.getContainerId(), testData.blobContent, testData.userMetadata, perBlobKey, cryptoService, kms, new CryptoJobMetricsTracker(routerMetrics.encryptJobMetrics), (EncryptJob.EncryptJobResult result, Exception exception) -> { encryptCallBackCount.countDown(); assertNotNull("Encrypted content should not be null", result.getEncryptedBlobContent()); assertNotNull("Encrypted user-metadata should not be null", result.getEncryptedUserMetadata()); assertNotNull("Encrypted key should not be null", result.getEncryptedKey()); result.getEncryptedBlobContent().release(); })); } else { encryptJobs.add( new EncryptJob(testData.blobId.getAccountId(), testData.blobId.getContainerId(), testData.blobContent, testData.userMetadata, perBlobKey, cryptoService, kms, new CryptoJobMetricsTracker(routerMetrics.encryptJobMetrics), (EncryptJob.EncryptJobResult result, Exception exception) -> { encryptCallBackCount.countDown(); if (exception == null) { assertNotNull("Encrypted content should not be null", result.getEncryptedBlobContent()); assertNotNull("Encrypted user-metadata should not be null", result.getEncryptedUserMetadata()); assertNotNull("Encrypted key should not be null", result.getEncryptedKey()); result.getEncryptedBlobContent().release(); } else { assertTrue("Exception cause should have been GeneralSecurityException", exception instanceof GeneralSecurityException); assertNull("Result should have been null", result); } })); } } TestBlobData probeData = getRandomBlob(referenceClusterMap); cryptoJobHandler.submitJob( new EncryptJob(probeData.blobId.getAccountId(), probeData.blobId.getContainerId(), probeData.blobContent, probeData.userMetadata, perBlobKey, cryptoService, kms, new CryptoJobMetricsTracker(routerMetrics.encryptJobMetrics), (EncryptJob.EncryptJobResult result, Exception exception) -> { result.getEncryptedBlobContent().release(); for (EncryptJob encryptJob : encryptJobs) { cryptoJobHandler.submitJob(encryptJob); } new Thread(new CloseCryptoHandler(cryptoJobHandler)).start(); })); awaitCountDownLatch(encryptCallBackCount, ENCRYPT_JOB_TYPE); }
void submitJob(CryptoJob cryptoJob) { if (enabled.get()) { scheduler.execute(cryptoJob); } }
CryptoJobHandler implements Closeable { void submitJob(CryptoJob cryptoJob) { if (enabled.get()) { scheduler.execute(cryptoJob); } } }
CryptoJobHandler implements Closeable { void submitJob(CryptoJob cryptoJob) { if (enabled.get()) { scheduler.execute(cryptoJob); } } CryptoJobHandler(int threadCount); }
CryptoJobHandler implements Closeable { void submitJob(CryptoJob cryptoJob) { if (enabled.get()) { scheduler.execute(cryptoJob); } } CryptoJobHandler(int threadCount); void close(); }
CryptoJobHandler implements Closeable { void submitJob(CryptoJob cryptoJob) { if (enabled.get()) { scheduler.execute(cryptoJob); } } CryptoJobHandler(int threadCount); void close();  }