@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, null,
          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, null, 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();
          }));
    }

    // wait for all encryption to complete
    awaitCountDownLatch(encryptCallBackCount, ENCRYPT_JOB_TYPE);

    // add special job that will close the thread. Add all the decrypt jobs to the queue before closing the thread.
    TestBlobData probeData = getRandomBlob(referenceClusterMap);
    cryptoJobHandler.submitJob(
        new EncryptJob(probeData.blobId.getAccountId(), probeData.blobId.getContainerId(), probeData.blobContent,
            probeData.userMetadata, perBlobKey, cryptoService, kms, null,
            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);
  }
Tests {@link CryptoJobHandler} for pending decrypt jobs callback after closing the thread
@throws InterruptedException
@throws GeneralSecurityException
ambry-router/src/test/java/com/github/ambry/router/CryptoJobHandlerTest.java