@Test
  public void testRouterMetadataCacheSimpleBlob() throws Exception {
    String blobId = null;
    try {
      Properties properties = getNonBlockingRouterProperties(localDcName);
      properties.setProperty("router.blob.metadata.cache.enabled", Boolean.toString(true));
      properties.setProperty("router.smallest.blob.for.metadata.cache", Long.toString(0));
      setRouterWithMetadataCache(properties, new AmbryCacheStats(1, 1, 1, 0));
      AmbryCacheWithStats ambryCacheWithStats = (AmbryCacheWithStats) router.getBlobMetadataCache();
      assertNotNull("Metadata cache must be created", ambryCacheWithStats);

      // put a simple blob and test it is absent in the metadata cache
      setOperationParams(PUT_CONTENT_SIZE, TTL_SECS);
      blobId = router.putBlob(putBlobProperties, putUserMetadata, putChannel, new PutBlobOptionsBuilder().build())
          .get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);

      // get the entire simple blob, and it's metadata must be absent from cache
      router.getBlob(blobId, new GetBlobOptionsBuilder().build()).get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      // Metadata for simple blob must be absent from metadata cache
      assertNull("Blob metadata must be absent in metadata cache", router.getBlobMetadataCache().getObject(blobId));

      // get the one byte from simple blob, and it's metadata must be absent from cache
      ByteRange range = ByteRanges.fromOffsetRange(1, 2);
      router.getBlob(blobId, new GetBlobOptionsBuilder().range(range).build())
          .get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      // Metadata for simple blob must be absent from metadata cache
      assertNull("Blob metadata must be absent in metadata cache", router.getBlobMetadataCache().getObject(blobId));

      // delete simple blob
      router.deleteBlob(blobId, null).get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
      // Metadata for simple blob must be absent from metadata cache
      assertNull("Blob metadata must be absent in metadata cache", router.getBlobMetadataCache().getObject(blobId));

      try {
        // get the one byte from simple blob and it's metadata must be absent from cache
        router.getBlob(blobId, new GetBlobOptionsBuilder().range(range).build())
            .get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
        fail("Blob must be deleted");
      } catch (Exception e) {
        RouterException r = (RouterException) e.getCause();
        Assert.assertEquals("BlobDeleted error is expected", RouterErrorCode.BlobDeleted, r.getErrorCode());
        // Metadata for simple blob must be absent from metadata cache
        assertNull("Blob metadata must be absent in metadata cache", router.getBlobMetadataCache().getObject(blobId));
      }
    } finally {
      router.close();
    }
  }
Response handling related tests for all operation managers.
ambry-router/src/test/java/com/github/ambry/router/NonBlockingRouterTest.java