@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, null,
                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, null,
                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);
                  }
                }));
      }
    }
    // add special job that will close the thread. Add all the encrypt 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 (EncryptJob encryptJob : encryptJobs) {
                cryptoJobHandler.submitJob(encryptJob);
              }
              new Thread(new CloseCryptoHandler(cryptoJobHandler)).start();
            }));

    awaitCountDownLatch(encryptCallBackCount, ENCRYPT_JOB_TYPE);
  }
Tests {@link CryptoJobHandler} for pending encrypt jobs callback after closing the thread
@throws InterruptedException
@throws GeneralSecurityException
ambry-router/src/test/java/com/github/ambry/router/CryptoJobHandlerTest.java