@Test public void testPendingDecryptJobs() throws InterruptedException, GeneralSecurityException { nettyByteBufLeakHelper.setDisabled(true); int testDataCount = 10; CountDownLatch encryptCallBackCount = new CountDownLatch(testDataCount); CountDownLatch decryptCallBackCount = new CountDownLatch(testDataCount); SecretKeySpec perBlobKey = kms.getRandomKey(); List<DecryptJob> decryptJobs = new CopyOnWriteArrayList<>(); for (int i = 0; i < testDataCount; i++) { TestBlobData testData = getRandomBlob(referenceClusterMap); cryptoJobHandler.submitJob(new EncryptJob(testData.blobId.getAccountId(), testData.blobId.getContainerId(), testData.blobContent.retainedDuplicate(), testData.userMetadata, perBlobKey, cryptoService, kms, new CryptoJobMetricsTracker(routerMetrics.encryptJobMetrics), (EncryptJob.EncryptJobResult encryptJobResult, Exception exception) -> { assertNotNull("Encrypted content should not be null", encryptJobResult.getEncryptedBlobContent()); assertNotNull("Encrypted key should not be null", encryptJobResult.getEncryptedKey()); decryptJobs.add(new DecryptJob(testData.blobId, encryptJobResult.getEncryptedKey(), encryptJobResult.getEncryptedBlobContent(), encryptJobResult.getEncryptedUserMetadata(), cryptoService, kms, new CryptoJobMetricsTracker(routerMetrics.decryptJobMetrics), (DecryptJob.DecryptJobResult decryptJobResult, Exception e) -> { if (e == null) { assertEquals("BlobId mismatch ", testData.blobId, decryptJobResult.getBlobId()); assertNull("Exception shouldn't have been thrown to decrypt contents for " + testData.blobId, exception); ByteBuf decryptedBlobContent = decryptJobResult.getDecryptedBlobContent(); assertNotNull("Decrypted contents should not be null", decryptedBlobContent); byte[] blobContent = new byte[testData.blobContent.readableBytes()]; testData.blobContent.readBytes(blobContent); byte[] decryptedBlobContentInBytes = new byte[decryptedBlobContent.readableBytes()]; decryptedBlobContent.readBytes(decryptedBlobContentInBytes); assertArrayEquals("Decrypted content and plain bytes should match", blobContent, decryptedBlobContentInBytes); decryptedBlobContent.release(); assertArrayEquals("Decrypted userMetadata and plain bytes should match", testData.userMetadata.array(), decryptJobResult.getDecryptedUserMetadata().array()); } else { assertNotNull("Exception should have been thrown to decrypt contents for " + testData.blobId, e); assertTrue("Exception cause should have been GeneralSecurityException", e instanceof GeneralSecurityException); assertNull("Result should have been null", decryptJobResult); } decryptCallBackCount.countDown(); })); encryptCallBackCount.countDown(); })); } awaitCountDownLatch(encryptCallBackCount, ENCRYPT_JOB_TYPE); 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 (DecryptJob decryptJob : decryptJobs) { cryptoJobHandler.submitJob(decryptJob); } new Thread(new CloseCryptoHandler(cryptoJobHandler)).start(); })); awaitCountDownLatch(decryptCallBackCount, DECRYPT_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();  }