1. Added support for BlobNotModifiedResponse 2. Fixed Connection leak…

… in CoordinatorBackedRouter (#286)

ambry-api/src/main/java/com.github.ambry/config/FrontendConfig.java

@@ -50,6 +50,13 @@ public class FrontendConfig {
   @Default("")
   public final List<String> frontendPathPrefixesToRemove;
 
+  /**
+   * Specifies the blob size in bytes beyond which chunked response will be sent for a getBlob() call
+   */
+  @Config("frontend.chunked.get.response.threshold.in.bytes")
+  @Default("8192")
+  public final Integer frontendChunkedGetResponseThresholdInBytes;
+
   public FrontendConfig(VerifiableProperties verifiableProperties) {
     frontendCacheValiditySeconds = verifiableProperties.getLong("frontend.cache.validity.seconds", 365 * 24 * 60 * 60);
     frontendIdConverterFactory = verifiableProperties
@@ -58,5 +65,7 @@ public FrontendConfig(VerifiableProperties verifiableProperties) {
         .getString("frontend.security.service.factory", "com.github.ambry.frontend.AmbrySecurityServiceFactory");
     frontendPathPrefixesToRemove =
         Arrays.asList(verifiableProperties.getString("frontend.path.prefixes.to.remove", "").split(","));
+    frontendChunkedGetResponseThresholdInBytes =
+        verifiableProperties.getInt("frontend.chunked.get.response.threshold.in.bytes", 8192);
   }
 }

ambry-api/src/main/java/com.github.ambry/rest/ResponseStatus.java

@@ -31,6 +31,11 @@ public enum ResponseStatus {
    */
   Accepted,
 
+  // 3xx
+  /**
+   * 304 Not Modified
+   */
+  NotModified,
   // 4xx
   /**
    * 400 - Request was not correct.

ambry-api/src/main/java/com.github.ambry/rest/RestUtils.java

@@ -18,8 +18,11 @@
 import com.github.ambry.utils.Utils;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -108,6 +111,11 @@ public static final class Headers {
      * Header to contain the Cookies
      */
     public final static String COOKIE = "Cookie";
+    /**
+     * Header to be set by the clients during a Get blob call to denote, that blob should be served only if the blob
+     * has been modified after the value set for this header.
+     */
+    public static final String IF_MODIFIED_SINCE = "If-Modified-Since";
   }
 
   /**
@@ -131,6 +139,7 @@ public static final class MultipartPost {
 
   private static final int Crc_Size = 8;
   private static final short UserMetadata_Version_V1 = 1;
+  public static final String HTTP_DATE_FORMAT = "EEE, dd MMM yyyy HH:mm:ss zzz";
 
   private static Logger logger = LoggerFactory.getLogger(RestUtils.class);
 
@@ -416,4 +425,32 @@ public static RestUtils.SubResource getBlobSubResource(RestRequest restRequest)
     }
     return subResource;
   }
+
+  /**
+   * Fetch time in ms for the {@code dateString} passed in, since epoch
+   * @param dateString the String representation of the date that needs to be parsed
+   * @return Time in ms since epoch. Note http time is kept in Seconds so last three digits will be 000.
+   *         Returns null if the {@code dateString} is not in the expected format or could not be parsed
+   */
+  public static Long getTimeFromDateString(String dateString) {
+    try {
+      SimpleDateFormat dateFormatter = new SimpleDateFormat(HTTP_DATE_FORMAT, Locale.US);
+      return dateFormatter.parse(dateString).getTime();
+    } catch (ParseException e) {
+      logger.warn("Could not parse milliseconds from an HTTP date header (" + dateString + ").");
+      return null;
+    }
+  }
+
+  /**
+   * Reduces the precision of a time in milliseconds to seconds precision. Result returned is in milliseconds with last
+   * three digits 000. Useful for comparing times kept in milliseconds that get converted to seconds and back (as is
+   * done with HTTP date format).
+   *
+   * @param ms time that needs to be parsed
+   * @return milliseconds with seconds precision (last three digits 000).
+   */
+  public static long toSecondsPrecisionInMs(long ms) {
+    return ms -= ms % 1000;
+  }
 }

ambry-api/src/test/java/com.github.ambry/rest/RestUtilsTest.java

@@ -20,11 +20,15 @@
 import java.net.URISyntaxException;
 import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
+import java.text.SimpleDateFormat;
 import java.util.Arrays;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Random;
+import java.util.TimeZone;
 import org.json.JSONException;
 import org.json.JSONObject;
 import org.junit.Test;
@@ -608,7 +612,7 @@ private void verifyUserMetadataConstructionSuccess(JSONObject headers, Map<Strin
       boolean keyFromInputMap = inputUserMetadata.containsKey(key);
       assertTrue("Key " + key + " not found in input user metadata", keyFromInputMap);
       assertTrue("Values didn't match for key " + key + ", value from input map value " + inputUserMetadata.get(key)
-              + ", and output map value " + userMetadataMap.get(key),
+          + ", and output map value " + userMetadataMap.get(key),
           inputUserMetadata.get(key).equals(userMetadataMap.get(key)));
     }
   }
@@ -666,4 +670,33 @@ public static void setUserMetadataHeaders(JSONObject headers, Map<String, String
       headers.put(key, userMetadata.get(key));
     }
   }
+
+  @Test
+  public void toSecondsPrecisionInMsTest() {
+    assertEquals(0, RestUtils.toSecondsPrecisionInMs(999));
+    assertEquals(1000, RestUtils.toSecondsPrecisionInMs(1000));
+    assertEquals(1000, RestUtils.toSecondsPrecisionInMs(1001));
+  }
+
+  @Test
+  public void getTimeFromDateStringTest() {
+    SimpleDateFormat dateFormatter = new SimpleDateFormat(RestUtils.HTTP_DATE_FORMAT, Locale.ENGLISH);
+    dateFormatter.setTimeZone(TimeZone.getTimeZone("GMT"));
+    long curTime = System.currentTimeMillis();
+    Date curDate = new Date(curTime);
+    String dateStr = dateFormatter.format(curDate);
+    long epochTime = RestUtils.getTimeFromDateString(dateStr);
+    long actualExpectedTime = (curTime / 1000L) * 1000;
+    // Note http time is kept in Seconds so last three digits will be 000
+    assertEquals("Time mismatch ", actualExpectedTime, epochTime);
+
+    dateFormatter = new SimpleDateFormat(RestUtils.HTTP_DATE_FORMAT, Locale.CHINA);
+    curTime = System.currentTimeMillis();
+    curDate = new Date(curTime);
+    dateStr = dateFormatter.format(curDate);
+    // any other locale is not accepted
+    assertEquals("Should have returned null", null, RestUtils.getTimeFromDateString(dateStr));
+
+    assertEquals("Should have returned null", null, RestUtils.getTimeFromDateString("abc"));
+  }
 }

ambry-frontend/src/main/java/com.github.ambry.frontend/AmbryBlobStorageService.java

@@ -612,11 +612,19 @@ public void onCompletion(final BlobInfo routerResult, Exception routerException)
             public void onCompletion(Void securityResult, Exception securityException) {
               securityCallbackTracker.markOperationEnd();
               ReadableStreamChannel response = null;
+              boolean blobNotModified = false;
               try {
                 if (securityException == null) {
                   if (subResource == null) {
-                    logger.trace("Forwarding GET after HEAD for {} to the router", blobId);
-                    router.getBlob(blobId, new GetCallback(restRequest, restResponseChannel));
+                    Long ifModifiedSinceMs = getIfModifiedSinceMs(restRequest);
+                    if (ifModifiedSinceMs != null
+                        && RestUtils.toSecondsPrecisionInMs(routerResult.getBlobProperties().getCreationTimeInMs())
+                        <= ifModifiedSinceMs) {
+                      blobNotModified = true;
+                    } else {
+                      logger.trace("Forwarding GET after HEAD for {} to the router", blobId);
+                      router.getBlob(blobId, new GetCallback(restRequest, restResponseChannel));
+                    }
                   } else {
                     Map<String, String> userMetadata = RestUtils.buildUserMetadata(routerResult.getUserMetadata());
                     if (userMetadata == null) {
@@ -636,7 +644,7 @@ public void onCompletion(Void securityResult, Exception securityException) {
                 securityException = e;
               } finally {
                 securityCallbackTracker.markCallbackProcessingEnd();
-                if (response != null || securityException != null) {
+                if (response != null || securityException != null || blobNotModified) {
                   submitResponse(restRequest, restResponseChannel, response, securityException);
                 }
               }
@@ -654,6 +662,18 @@ public void onCompletion(Void securityResult, Exception securityException) {
       }
     }
 
+    /**
+     * Fetches the {@link RestUtils.Headers#IF_MODIFIED_SINCE} value in epoch time if present
+     * @param restRequest the {@link RestRequest} that needs to be parsed
+     * @return the {@link RestUtils.Headers#IF_MODIFIED_SINCE} value in epoch time if present
+     */
+    private Long getIfModifiedSinceMs(RestRequest restRequest) {
+      if (restRequest.getArgs().get(RestUtils.Headers.IF_MODIFIED_SINCE) != null) {
+        return RestUtils.getTimeFromDateString((String) restRequest.getArgs().get(RestUtils.Headers.IF_MODIFIED_SINCE));
+      }
+      return null;
+    }
+
     /**
      * Sets the blob ID that should be used for {@link Router#getBlob(String, Callback)}.
      * @param blobId the blob ID that should be used for {@link Router#getBlob(String, Callback)}.

ambry-frontend/src/main/java/com.github.ambry.frontend/AmbrySecurityService.java

@@ -40,11 +40,11 @@
 class AmbrySecurityService implements SecurityService {
 
   private boolean isOpen;
-  private final long cacheValidityInSecs;
+  private final FrontendConfig frontendConfig;
   private final FrontendMetrics frontendMetrics;
 
   public AmbrySecurityService(FrontendConfig frontendConfig, FrontendMetrics frontendMetrics) {
-    cacheValidityInSecs = frontendConfig.frontendCacheValiditySeconds;
+    this.frontendConfig = frontendConfig;
     this.frontendMetrics = frontendMetrics;
     isOpen = true;
   }
@@ -86,16 +86,31 @@ public Future<Void> processResponse(RestRequest restRequest, RestResponseChannel
       try {
         responseChannel.setStatus(ResponseStatus.Ok);
         responseChannel.setHeader(RestUtils.Headers.DATE, new GregorianCalendar().getTime());
-        responseChannel
-            .setHeader(RestUtils.Headers.LAST_MODIFIED, new Date(blobInfo.getBlobProperties().getCreationTimeInMs()));
         if (restRequest.getRestMethod() == RestMethod.HEAD) {
+          responseChannel
+              .setHeader(RestUtils.Headers.LAST_MODIFIED, new Date(blobInfo.getBlobProperties().getCreationTimeInMs()));
           setHeadResponseHeaders(blobInfo, responseChannel);
         } else if (restRequest.getRestMethod() == RestMethod.GET) {
           RestUtils.SubResource subResource = RestUtils.getBlobSubResource(restRequest);
           if (subResource == null) {
-            setGetBlobResponseHeaders(responseChannel, blobInfo);
+            Long ifModifiedSinceMs = getIfModifiedSinceMs(restRequest);
+            if (ifModifiedSinceMs != null
+                && RestUtils.toSecondsPrecisionInMs(blobInfo.getBlobProperties().getCreationTimeInMs())
+                <= ifModifiedSinceMs) {
+              responseChannel.setStatus(ResponseStatus.NotModified);
+              responseChannel.setHeader(RestUtils.Headers.CONTENT_LENGTH, 0);
+            } else {
+              responseChannel.setHeader(RestUtils.Headers.LAST_MODIFIED,
+                  new Date(blobInfo.getBlobProperties().getCreationTimeInMs()));
+              setGetBlobResponseHeaders(responseChannel, blobInfo);
+            }
           } else if (subResource == RestUtils.SubResource.BlobInfo) {
+            responseChannel.setHeader(RestUtils.Headers.LAST_MODIFIED,
+                new Date(blobInfo.getBlobProperties().getCreationTimeInMs()));
             setBlobPropertiesHeaders(blobInfo.getBlobProperties(), responseChannel);
+          } else if (subResource == RestUtils.SubResource.UserMetadata) {
+            responseChannel.setHeader(RestUtils.Headers.LAST_MODIFIED,
+                new Date(blobInfo.getBlobProperties().getCreationTimeInMs()));
           }
         }
       } catch (RestServiceException e) {
@@ -115,6 +130,18 @@ public void close() {
     isOpen = false;
   }
 
+  /**
+   * Fetches the {@link RestUtils.Headers#IF_MODIFIED_SINCE} value in epoch time if present
+   * @param restRequest the {@link RestRequest} that needs to be parsed
+   * @return the {@link RestUtils.Headers#IF_MODIFIED_SINCE} value in epoch time if present
+   */
+  private Long getIfModifiedSinceMs(RestRequest restRequest) {
+    if (restRequest.getArgs().get(RestUtils.Headers.IF_MODIFIED_SINCE) != null) {
+      return RestUtils.getTimeFromDateString((String) restRequest.getArgs().get(RestUtils.Headers.IF_MODIFIED_SINCE));
+    }
+    return null;
+  }
+
   /**
    * Sets the required headers in the HEAD response.
    * @param blobInfo the {@link BlobInfo} to refer to while setting headers.
@@ -140,6 +167,9 @@ private void setGetBlobResponseHeaders(RestResponseChannel restResponseChannel,
       throws RestServiceException {
     BlobProperties blobProperties = blobInfo.getBlobProperties();
     restResponseChannel.setHeader(RestUtils.Headers.BLOB_SIZE, blobProperties.getBlobSize());
+    if (blobProperties.getBlobSize() < frontendConfig.frontendChunkedGetResponseThresholdInBytes) {
+      restResponseChannel.setHeader(RestUtils.Headers.CONTENT_LENGTH, blobProperties.getBlobSize());
+    }
     if (blobProperties.getContentType() != null) {
       restResponseChannel.setHeader(RestUtils.Headers.CONTENT_TYPE, blobProperties.getContentType());
       // Ensure browsers do not execute html with embedded exploits.
@@ -153,8 +183,9 @@ private void setGetBlobResponseHeaders(RestResponseChannel restResponseChannel,
       restResponseChannel.setHeader(RestUtils.Headers.PRAGMA, "no-cache");
     } else {
       restResponseChannel.setHeader(RestUtils.Headers.EXPIRES,
-          new Date(System.currentTimeMillis() + cacheValidityInSecs * Time.MsPerSec));
-      restResponseChannel.setHeader(RestUtils.Headers.CACHE_CONTROL, "max-age=" + cacheValidityInSecs);
+          new Date(System.currentTimeMillis() + frontendConfig.frontendCacheValiditySeconds * Time.MsPerSec));
+      restResponseChannel
+          .setHeader(RestUtils.Headers.CACHE_CONTROL, "max-age=" + frontendConfig.frontendCacheValiditySeconds);
     }
   }
 

ambry-frontend/src/test/java/com.github.ambry.frontend/AmbryBlobStorageServiceTest.java

@@ -52,13 +52,17 @@
 import java.lang.reflect.Method;
 import java.net.URISyntaxException;
 import java.nio.ByteBuffer;
+import java.text.SimpleDateFormat;
 import java.util.ArrayList;
+import java.util.Date;
 import java.util.HashMap;
 import java.util.Iterator;
 import java.util.LinkedList;
 import java.util.List;
+import java.util.Locale;
 import java.util.Map;
 import java.util.Properties;
+import java.util.TimeZone;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
@@ -340,6 +344,7 @@ public void postGetHeadDeleteTest()
     RestUtilsTest.setUserMetadataHeaders(headers, userMetadata);
     String blobId = postBlobAndVerify(headers, content);
     getBlobAndVerify(blobId, headers, content);
+    getNotModifiedBlobAndVerify(blobId);
     getUserMetadataAndVerify(blobId, headers);
     getBlobInfoAndVerify(blobId, headers);
     getHeadAndVerify(blobId, headers);
@@ -697,6 +702,31 @@ public void getBlobAndVerify(String blobId, JSONObject expectedHeaders, ByteBuff
         restResponseChannel.getResponseBody());
   }
 
+  /**
+   * Gets the blob with blob ID {@code blobId} and verifies that the blob is not returned as blob is not modified
+   * @param blobId the blob ID of the blob to GET.
+   * @throws Exception
+   */
+  public void getNotModifiedBlobAndVerify(String blobId)
+      throws Exception {
+    JSONObject headers = new JSONObject();
+    SimpleDateFormat dateFormat = new SimpleDateFormat(RestUtils.HTTP_DATE_FORMAT, Locale.ENGLISH);
+    dateFormat.setTimeZone(TimeZone.getTimeZone("GMT"));
+    Date date = new Date(System.currentTimeMillis());
+    String dateStr = dateFormat.format(date);
+    headers.put(RestUtils.Headers.IF_MODIFIED_SINCE, dateStr);
+    RestRequest restRequest = createRestRequest(RestMethod.GET, blobId, headers, null);
+    MockRestResponseChannel restResponseChannel = new MockRestResponseChannel();
+    doOperation(restRequest, restResponseChannel);
+    assertEquals("Unexpected response status", ResponseStatus.NotModified, restResponseChannel.getResponseStatus());
+    assertTrue("No Date header", restResponseChannel.getHeader(RestUtils.Headers.DATE) != null);
+    assertNull("No Last-Modified header expected", restResponseChannel.getHeader("Last-Modified"));
+    assertNull(RestUtils.Headers.BLOB_SIZE + " should have been null ",
+        restResponseChannel.getHeader(RestUtils.Headers.BLOB_SIZE));
+    assertNull("Content-Type should have been null", restResponseChannel.getHeader(RestUtils.Headers.CONTENT_TYPE));
+    assertEquals("No content expected as blob is not modified", 0, restResponseChannel.getResponseBody().length);
+  }
+
   /**
    * Gets the user metadata of the blob with blob ID {@code blobId} and verifies them against what is expected.
    * @param blobId the blob ID of the blob to HEAD.