diff --git a/ambry-api/src/main/java/com/github/ambry/rest/RestUtils.java b/ambry-api/src/main/java/com/github/ambry/rest/RestUtils.java index d53ffb4fa6..8627092767 100644 --- a/ambry-api/src/main/java/com/github/ambry/rest/RestUtils.java +++ b/ambry-api/src/main/java/com/github/ambry/rest/RestUtils.java @@ -514,6 +514,11 @@ public static final class InternalKeys { */ public static final String REQUEST_PATH = KEY_PREFIX + "request-path"; + /** + * Boolean context for consuming a server-generated upload ID. Cleared after successful conversion or TTL promotion. + */ + public static final String BLOB_ID_IS_SERVER_GENERATED = KEY_PREFIX + "blob-id-is-server-generated"; + /** * The internal header to determine if the delete request is coming from a dataset deletion. */ diff --git a/ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendMetrics.java b/ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendMetrics.java index 206654ebd7..1915893c9a 100644 --- a/ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendMetrics.java +++ b/ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendMetrics.java @@ -255,6 +255,7 @@ public class FrontendMetrics { // Errors // FrontendRestRequestService public final Counter responseSubmissionError; + public final Counter generatedBlobIdError; public final Counter resourceReleaseError; public final Counter ttlTooLargeError; public final Counter ttlNotCompliantError; @@ -707,6 +708,8 @@ public FrontendMetrics(MetricRegistry metricRegistry, FrontendConfig frontendCon // FrontendRestRequestService responseSubmissionError = metricRegistry.counter(MetricRegistry.name(FrontendRestRequestService.class, "ResponseSubmissionError")); + generatedBlobIdError = + metricRegistry.counter(MetricRegistry.name(FrontendRestRequestService.class, "GeneratedBlobIdError")); resourceReleaseError = metricRegistry.counter(MetricRegistry.name(FrontendRestRequestService.class, "ResourceReleaseError")); // DeleteCallback diff --git a/ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendRestRequestService.java b/ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendRestRequestService.java index 66904e689b..906c5265fd 100644 --- a/ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendRestRequestService.java +++ b/ambry-frontend/src/main/java/com/github/ambry/frontend/FrontendRestRequestService.java @@ -561,6 +561,7 @@ public void handleOptions(RestRequest restRequest, RestResponseChannel restRespo /** * Submits the response and {@code responseBody} (and any {@code exception})for the {@code restRequest} to the * {@code responseHandler}. + * Counts invalid IDs encountered while consuming server-generated upload results. * @param restRequest the {@link RestRequest} for which a response is ready. * @param restResponseChannel the {@link RestResponseChannel} over which the response can be sent. * @param responseBody the body of the response in the form of a {@link ReadableStreamChannel}. @@ -575,8 +576,12 @@ void submitResponse(RestRequest restRequest, RestResponseChannel restResponseCha restResponseChannel.setHeader(TrackingHeaders.FRONTEND_NAME, hostname); } if (exception instanceof RouterException) { - exception = new RestServiceException(exception, - RestServiceErrorCode.getRestServiceErrorCode(((RouterException) exception).getErrorCode())); + RouterErrorCode routerErrorCode = ((RouterException) exception).getErrorCode(); + if (routerErrorCode == RouterErrorCode.InvalidBlobId + && Boolean.TRUE.equals(restRequest.getArgs().get(BLOB_ID_IS_SERVER_GENERATED))) { + frontendMetrics.generatedBlobIdError.inc(); + } + exception = new RestServiceException(exception, RestServiceErrorCode.getRestServiceErrorCode(routerErrorCode)); } responseHandler.handleResponse(restRequest, restResponseChannel, responseBody, exception); } catch (Exception e) { diff --git a/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobPutHandler.java b/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobPutHandler.java index ba5ed9fc96..8e8bcbfda2 100644 --- a/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobPutHandler.java +++ b/ambry-frontend/src/main/java/com/github/ambry/frontend/NamedBlobPutHandler.java @@ -244,6 +244,7 @@ private Callback routerPutBlobCallback(BlobInfo blobInfo) { restRequest.getMetricsTracker().setBytesTransferred(restRequest.getBytesReceived()); restResponseChannel.setHeader(RestUtils.Headers.BLOB_SIZE, restRequest.getBlobBytesReceived()); restResponseChannel.setHeader(RestUtils.Headers.LOCATION, blobId); + restRequest.setArg(BLOB_ID_IS_SERVER_GENERATED, true); String blobIdClean = stripPrefixAndExtension(blobId); if (blobInfo.getBlobProperties().getTimeToLiveInSeconds() == Utils.Infinite_Time) { // Do ttl update with retryExecutor. Use the blob ID returned from the router instead of the converted ID @@ -254,6 +255,7 @@ private Callback routerPutBlobCallback(BlobInfo blobInfo) { QuotaUtils.buildQuotaChargeCallback(restRequest, quotaManager, false)), this::isRetriable, routerTtlUpdateCallbackForPut(blobInfo)); } else { + restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED); if (RestUtils.isDatasetVersionQueryEnabled(restRequest.getArgs())) { //Make sure to process response after delete finished updateVersionStateAndDeleteDatasetVersionOutOfRetentionCount( @@ -291,6 +293,7 @@ private Callback fetchStitchRequestBodyCallback(RetainingAsyncWritableChan */ private Callback routerStitchBlobCallback(BlobInfo blobInfo, BlobProperties propertiesForRouterUpload) { return buildCallback(frontendMetrics.putRouterStitchBlobMetrics, convertedBlobId -> { + restRequest.setArg(BLOB_ID_IS_SERVER_GENERATED, true); // The actual blob size is now present in the instance of BlobProperties passed to the router.stitchBlob(). // Update it in the BlobInfo so that IdConverter can add it to the named blob DB blobInfo.getBlobProperties().setBlobSize(propertiesForRouterUpload.getBlobSize()); @@ -305,6 +308,7 @@ private Callback routerStitchBlobCallback(BlobInfo blobInfo, BlobPropert QuotaUtils.buildQuotaChargeCallback(restRequest, quotaManager, false)), this::isRetriable, routerTtlUpdateCallbackForStitch(blobInfo)); } else { + restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED); if (RestUtils.isDatasetVersionQueryEnabled(restRequest.getArgs())) { //Make sure to process response after delete finished updateVersionStateAndDeleteDatasetVersionOutOfRetentionCount( @@ -335,6 +339,7 @@ private boolean isRetriable(Throwable throwable) { */ private Callback routerTtlUpdateCallbackForPut(BlobInfo blobInfo) { return buildCallback(frontendMetrics.updateBlobTtlRouterMetrics, convertedBlobId -> { + restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED); if (RestUtils.isDatasetVersionQueryEnabled(restRequest.getArgs())) { //Make sure to process response after delete finished updateVersionStateAndDeleteDatasetVersionOutOfRetentionCount( @@ -354,6 +359,7 @@ private Callback routerTtlUpdateCallbackForPut(BlobInfo blobInfo) { */ private Callback routerTtlUpdateCallbackForStitch(BlobInfo blobInfo) { return buildCallback(frontendMetrics.updateBlobTtlRouterMetrics, convertedBlobId -> { + restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED); if (RestUtils.isDatasetVersionQueryEnabled(restRequest.getArgs())) { //Make sure to process response after delete finished updateVersionStateAndDeleteDatasetVersionOutOfRetentionCount( diff --git a/ambry-frontend/src/main/java/com/github/ambry/frontend/PostBlobHandler.java b/ambry-frontend/src/main/java/com/github/ambry/frontend/PostBlobHandler.java index 08a8e4f1f1..a6398b8677 100644 --- a/ambry-frontend/src/main/java/com/github/ambry/frontend/PostBlobHandler.java +++ b/ambry-frontend/src/main/java/com/github/ambry/frontend/PostBlobHandler.java @@ -223,9 +223,10 @@ private Callback fetchStitchRequestBodyCallback(RetainingAsyncWritableChan * @return a {@link Callback} to be used with {@link Router#putBlob}. */ private Callback routerStitchBlobCallback(BlobInfo blobInfo) { - return buildCallback(frontendMetrics.postRouterStitchBlobMetrics, - blobId -> idConverter.convert(restRequest, blobId, blobInfo.getBlobProperties(), idConverterCallback(blobInfo)), uri, LOGGER, - finalCallback); + return buildCallback(frontendMetrics.postRouterStitchBlobMetrics, blobId -> { + restRequest.setArg(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED, true); + idConverter.convert(restRequest, blobId, blobInfo.getBlobProperties(), idConverterCallback(blobInfo)); + }, uri, LOGGER, finalCallback); } /** @@ -238,6 +239,7 @@ private Callback routerPutBlobCallback(BlobInfo blobInfo) { return buildCallback(frontendMetrics.postRouterPutBlobMetrics, blobId -> { setSignedIdMetadataAndBlobSize(blobInfo.getBlobProperties()); restRequest.getMetricsTracker().setBytesTransferred(restRequest.getBytesReceived()); + restRequest.setArg(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED, true); idConverter.convert(restRequest, blobId, idConverterCallback(blobInfo)); }, uri, LOGGER, finalCallback); } @@ -250,6 +252,7 @@ private Callback routerPutBlobCallback(BlobInfo blobInfo) { */ private Callback idConverterCallback(BlobInfo blobInfo) { return buildCallback(frontendMetrics.postIdConversionMetrics, convertedBlobId -> { + restRequest.removeArg(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED); restResponseChannel.setHeader(RestUtils.Headers.LOCATION, convertedBlobId); securityService.processResponse(restRequest, restResponseChannel, blobInfo, securityProcessResponseCallback()); }, uri, LOGGER, finalCallback); diff --git a/ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartCompleteUploadHandler.java b/ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartCompleteUploadHandler.java index 5ab7a831dc..54cd619df5 100644 --- a/ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartCompleteUploadHandler.java +++ b/ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartCompleteUploadHandler.java @@ -241,6 +241,7 @@ private Callback fetchStitchRequestBodyCallback(RetainingAsyncWritableChan private Callback routerStitchBlobCallback(BlobInfo blobInfo, BlobProperties propertiesPassedInRouterUpload) { return buildCallback(frontendMetrics.putRouterStitchBlobMetrics, blobId -> { + restRequest.setArg(BLOB_ID_IS_SERVER_GENERATED, true); // The actual blob size is now present in the instance of BlobProperties passed to the router.stitchBlob(). // Update it in the BlobInfo so that IdConverter can add it to the named blob DB blobInfo.getBlobProperties().setBlobSize(propertiesPassedInRouterUpload.getBlobSize()); @@ -277,6 +278,7 @@ private Callback idConverterCallback(BlobInfo blobInfo, String blobId) { */ private Callback routerTtlUpdateCallback(BlobInfo blobInfo, String blobId) { return buildCallback(frontendMetrics.updateBlobTtlRouterMetrics, convertedBlobId -> { + restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED); // Set the named blob state to be 'READY' after the Ttl update succeed if (!restRequest.getArgs().containsKey(NAMED_BLOB_VERSION)) { throw new RestServiceException( diff --git a/ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartUploadPartHandler.java b/ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartUploadPartHandler.java index 4b2bfd55f7..c2cf9a9ba7 100644 --- a/ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartUploadPartHandler.java +++ b/ambry-frontend/src/main/java/com/github/ambry/frontend/s3/S3MultipartUploadPartHandler.java @@ -188,6 +188,7 @@ private Callback securityPostProcessRequestCallback(BlobInfo blobInfo) { */ private Callback routerPutBlobCallback(BlobInfo blobInfo) { return buildCallback(frontendMetrics.putRouterPutBlobMetrics, blobId -> { + restRequest.setArg(BLOB_ID_IS_SERVER_GENERATED, true); restResponseChannel.setHeader(RestUtils.Headers.BLOB_SIZE, restRequest.getBlobBytesReceived()); // TODO [S3] Make changes to sign ETags. Currently they are sent as shown below. // ETag: {"chunks":[{"blob":"AAYQAQBlAAgAAQAAAAAAAAAAw6UGCoNgS8KGgV-SGXAMdQ","size":4194304}, @@ -196,6 +197,7 @@ private Callback routerPutBlobCallback(BlobInfo blobInfo) { S3MultipartETag etag = new S3MultipartETag(putBlobMetaInfo.getOrderedChunkIdSizeList()); String eTagStr = S3MultipartETag.serialize(etag); restResponseChannel.setHeader(RestUtils.Headers.LOCATION, eTagStr); + restRequest.removeArg(BLOB_ID_IS_SERVER_GENERATED); securityService.processResponse(restRequest, restResponseChannel, blobInfo, securityProcessResponseCallback()); }, uri, logger, finalCallback); } diff --git a/ambry-frontend/src/test/java/com/github/ambry/frontend/FrontendRestRequestServiceTest.java b/ambry-frontend/src/test/java/com/github/ambry/frontend/FrontendRestRequestServiceTest.java index a4f1553316..77a27eb94f 100644 --- a/ambry-frontend/src/test/java/com/github/ambry/frontend/FrontendRestRequestServiceTest.java +++ b/ambry-frontend/src/test/java/com/github/ambry/frontend/FrontendRestRequestServiceTest.java @@ -43,6 +43,7 @@ import com.github.ambry.config.QuotaConfig; import com.github.ambry.config.RouterConfig; import com.github.ambry.config.VerifiableProperties; +import com.github.ambry.frontend.s3.S3MultipartETag; import com.github.ambry.messageformat.BlobInfo; import com.github.ambry.messageformat.BlobProperties; import com.github.ambry.named.DeleteResult; @@ -134,6 +135,7 @@ import org.junit.Assert; import org.junit.Test; import org.mockito.Mockito; +import org.mockito.stubbing.Answer; import static com.github.ambry.rest.RestUtils.*; import static com.github.ambry.rest.RestUtils.Headers.*; @@ -2243,6 +2245,180 @@ public void routerExceptionPipelineTest() throws Exception { doRouterExceptionPipelineTest(testRouter, exceptionMsg + " Error: " + RouterErrorCode.UnexpectedInternalError); } + @Test + public void testGeneratedIdResponseMapping() throws Exception { + assertSame(frontendMetrics.generatedBlobIdError, + metricRegistry.getCounters().get(MetricRegistry.name(FrontendRestRequestService.class, "GeneratedBlobIdError"))); + for (Object context : new Object[]{null, false, true, "true"}) { + for (RouterErrorCode code : RouterErrorCode.values()) { + try (RestRequest request = createRestRequest(RestMethod.POST, "/", null, null)) { + if (context != null) { + request.setArg(InternalKeys.BLOB_ID_IS_SERVER_GENERATED, context); + } + RouterException error = new RouterException("router failure", code); + MockRestResponseChannel response = spy(new MockRestResponseChannel()); + long countBefore = frontendMetrics.generatedBlobIdError.getCount(); + frontendRestRequestService.submitResponse(request, response, null, error); + boolean generatedIdError = Boolean.TRUE.equals(context) && code == RouterErrorCode.InvalidBlobId; + assertEquals(ResponseStatus.getResponseStatus(RestServiceErrorCode.getRestServiceErrorCode(code)), + response.getStatus()); + assertSame(error, response.getException().getCause()); + assertEquals(countBefore + (generatedIdError ? 1 : 0), frontendMetrics.generatedBlobIdError.getCount()); + assertEquals(0, frontendMetrics.responseSubmissionError.getCount()); + verify(response).onResponseComplete(any()); + } + } + } + } + + @Test + public void testGeneratedIdCounterIgnoresNonRouterFailures() throws Exception { + for (Exception error : new Exception[]{null, new RestServiceException("invalid input", RestServiceErrorCode.BadRequest), + new RuntimeException("failure")}) { + try (RestRequest request = createRestRequest(RestMethod.POST, "/", null, null)) { + request.setArg(InternalKeys.BLOB_ID_IS_SERVER_GENERATED, true); + MockRestResponseChannel response = new MockRestResponseChannel(); + long countBefore = frontendMetrics.generatedBlobIdError.getCount(); + frontendRestRequestService.submitResponse(request, response, null, error); + assertSame(error, response.getException()); + assertEquals(countBefore, frontendMetrics.generatedBlobIdError.getCount()); + } + } + } + + @Test + public void testUploadBlobIdErrorClassification() throws Exception { + refContainer = new ContainerBuilder(refContainer).setNamedBlobMode(Container.NamedBlobMode.OPTIONAL) + .setTtlRequired(false).build(); + accountService.updateAccounts( + Collections.singletonList(new AccountBuilder(refAccount).addOrUpdateContainer(refContainer).build())); + FrontendTestIdConverterFactory converterFactory = new FrontendTestIdConverterFactory(); + converterFactory.returnInputIfTranslationNull = true; + idConverterFactory = converterFactory; + router = spy(router); + doReturn(converterFactory.getIdConverter()).when(router).getIdConverter(); + frontendRestRequestService = spy(getFrontendRestRequestService()); + frontendRestRequestService.start(); + String namedPath = "/named/" + refAccount.getName() + "/" + refContainer.getName() + "/blob"; + String s3Path = namedPath.replace("/named/", "/s3/"); + String chunkId = new BlobId(blobIdVersion, BlobId.BlobIdType.NATIVE, ClusterMap.UNKNOWN_DATACENTER_ID, + refAccount.getId(), refContainer.getId(), referenceBlobId.getPartition(), false, + BlobId.BlobDataType.DATACHUNK).getID(); + Map metadata = new HashMap<>(); + metadata.put(SESSION, "session"); + metadata.put(BLOB_SIZE, "1"); + metadata.put(PostBlobHandler.EXPIRATION_TIME_MS_KEY, Long.toString(Long.MAX_VALUE)); + metadata.put(RESERVED_METADATA_ID, referenceBlobIdStr); + String stitchBody = StitchRequestSerDe.toJson( + Collections.singletonList(idSigningService.getSignedId(chunkId, metadata))).toString(); + String etag = S3MultipartETag.serialize(new S3MultipartETag(Collections.singletonList(new Pair<>(chunkId, 1L)))); + String completeBody = "1" + etag + + ""; + Object[][] requests = { + {RestMethod.POST, "/", false, ""}, + {RestMethod.PUT, namedPath, false, ""}, + {RestMethod.PUT, s3Path, false, ""}, + {RestMethod.PUT, s3Path + "?uploadId=session&partNumber=1", false, ""}, + {RestMethod.POST, "/stitch", true, stitchBody}, + {RestMethod.PUT, namedPath, true, stitchBody}, + {RestMethod.POST, s3Path + "?uploadId=session", true, completeBody} + }; + for (Object[] requestCase : requests) { + RestMethod method = (RestMethod) requestCase[0]; + String uri = (String) requestCase[1]; + boolean stitch = (boolean) requestCase[2]; + for (String stage : new String[]{"router", "conversion", "ttl"}) { + if (stage.equals("conversion") && method != RestMethod.POST + || stage.equals("ttl") && (uri.equals("/") || uri.equals("/stitch") || uri.contains("partNumber"))) { + continue; + } + for (Exception failure : new Exception[]{ + new RouterException("invalid ID", RouterErrorCode.InvalidBlobId), + new RouterException("too large", RouterErrorCode.BlobTooLarge), + new RestServiceException("invalid input", RestServiceErrorCode.BadRequest)}) { + converterFactory.exceptionToReturn = stage.equals("conversion") ? failure : null; + Answer> upload = invocation -> { + Callback callback = invocation.getArgument(5); + callback.onCompletion("invalid-generated-id", stage.equals("router") ? failure : null); + return CompletableFuture.completedFuture("invalid-generated-id"); + }; + doAnswer(upload).when(router).putBlob(any(), any(), any(), any(), any(), any(), any()); + doAnswer(upload).when(router).stitchBlob(any(), any(), any(), any(), any(), any(), any()); + doAnswer(invocation -> { + assertEquals("invalid-generated-id", invocation.getArgument(1)); + Callback callback = invocation.getArgument(4); + callback.onCompletion(null, failure); + return CompletableFuture.completedFuture(null); + }).when(router).updateBlobTtl(any(), anyString(), any(), anyLong(), any(), any()); + boolean parseGeneratedId = stage.equals("ttl") && failure instanceof RouterException + && ((RouterException) failure).getErrorCode() == RouterErrorCode.InvalidBlobId; + if (parseGeneratedId) { + doCallRealMethod().when(router).updateBlobTtl(any(), anyString(), any(), anyLong(), any(), any()); + } + JSONObject headers = new JSONObject(); + setAmbryHeadersForPut(headers, Utils.Infinite_Time, !refContainer.isCacheable(), "test", + "application/octet-stream", "test", refAccount.getName(), refContainer.getName(), null); + if (stitch && method == RestMethod.PUT) { + headers.put(UPLOAD_NAMED_BLOB_MODE, STITCH); + } + byte[] body = ((String) requestCase[3]).getBytes(StandardCharsets.UTF_8); + clearInvocations(router); + try (RestRequest request = createRestRequest(method, uri, headers, + new LinkedList<>(Arrays.asList(ByteBuffer.wrap(body), null)))) { + long countBefore = frontendMetrics.generatedBlobIdError.getCount(); + RestServiceErrorCode expected; + if (failure instanceof RouterException + && ((RouterException) failure).getErrorCode() == RouterErrorCode.BlobTooLarge) { + expected = RestServiceErrorCode.RequestTooLarge; + } else { + expected = RestServiceErrorCode.BadRequest; + } + boolean generatedIdError = !stage.equals("router") && failure instanceof RouterException + && ((RouterException) failure).getErrorCode() == RouterErrorCode.InvalidBlobId; + MockRestResponseChannel response = spy(new MockRestResponseChannel()); + assertException(RestServiceException.class, () -> doOperation(request, response), + e -> assertEquals(uri + ":" + stage, expected, e.getErrorCode())); + assertEquals(uri + ":" + stage, ResponseStatus.getResponseStatus(expected), response.getStatus()); + assertEquals(countBefore + (generatedIdError ? 1 : 0), frontendMetrics.generatedBlobIdError.getCount()); + verify(response).onResponseComplete(any()); + if (parseGeneratedId) { + assertEquals(RouterErrorCode.InvalidBlobId, + ((RouterException) response.getException().getCause()).getErrorCode()); + } else { + assertSame(failure, failure instanceof RouterException ? response.getException().getCause() + : response.getException()); + } + verify(frontendRestRequestService).submitResponse(request, response, null, + parseGeneratedId ? (RouterException) response.getException().getCause() : failure); + verify(router, times(stage.equals("ttl") ? 1 : 0)) + .updateBlobTtl(any(), anyString(), any(), anyLong(), any(), any()); + } + } + } + } + } + + @Test + public void testClientBlobIdErrorsRemainBadRequest() throws Exception { + Properties properties = new Properties(); + properties.setProperty(InMemoryRouter.OPERATION_THROW_ROUTER_EXCEPTION, RouterErrorCode.InvalidBlobId.name()); + router.setVerifiableProperties(new VerifiableProperties(properties)); + for (String blobId : new String[]{"invalid-client-id", referenceBlobIdStr}) { + for (String operation : new String[]{"GET", "HEAD", "DELETE", Operations.UPDATE_TTL, Operations.UNDELETE}) { + boolean put = operation.equals(Operations.UPDATE_TTL) || operation.equals(Operations.UNDELETE); + JSONObject headers = new JSONObject().put(BLOB_ID, blobId).put(SERVICE_ID, "test"); + headers.put(InternalKeys.BLOB_ID_IS_SERVER_GENERATED, "true"); + try (RestRequest request = createRestRequest(put ? RestMethod.PUT : RestMethod.valueOf(operation), + put ? "/" + operation : "/" + blobId, headers, null)) { + long countBefore = frontendMetrics.generatedBlobIdError.getCount(); + assertEquals(ResponseStatus.BadRequest, + verifyOperationFailure(request, RestServiceErrorCode.BadRequest).getStatus()); + assertEquals(countBefore, frontendMetrics.generatedBlobIdError.getCount()); + } + } + } + } + /** * Test that GET operations fail with the expected error code when a bad range header is provided. * @throws Exception diff --git a/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobPutHandlerTest.java b/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobPutHandlerTest.java index 3416042c08..958a9d3456 100644 --- a/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobPutHandlerTest.java +++ b/ambry-frontend/src/test/java/com/github/ambry/frontend/NamedBlobPutHandlerTest.java @@ -563,6 +563,7 @@ private void stitchBlobAndVerify(byte[] requestBody, List expectedSti namedBlobPutHandler.handle(request, restResponseChannel, future::done); if (errorChecker == null) { future.get(TIMEOUT_SECS, TimeUnit.SECONDS); + assertNull(request.getArgs().get(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED)); // in id converter, the callback is invoked before the lastConvertedId is set, we need to wait // for a while before we can check the lastConvertedId. Thread.sleep(100); @@ -731,6 +732,7 @@ private void putBlobAndVerify(ThrowingConsumer errorChecker, namedBlobPutHandler.handle(request, restResponseChannel, future::done); if (errorChecker == null) { future.get(TIMEOUT_SECS, TimeUnit.SECONDS); + assertNull(request.getArgs().get(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED)); // in id converter, the callback is invoked before the lastConvertedId is set, we need to wait // for a while before we can check the lastConvertedId. Thread.sleep(100); diff --git a/ambry-frontend/src/test/java/com/github/ambry/frontend/PostBlobHandlerTest.java b/ambry-frontend/src/test/java/com/github/ambry/frontend/PostBlobHandlerTest.java index 50be0d0eeb..03041889ad 100644 --- a/ambry-frontend/src/test/java/com/github/ambry/frontend/PostBlobHandlerTest.java +++ b/ambry-frontend/src/test/java/com/github/ambry/frontend/PostBlobHandlerTest.java @@ -536,6 +536,7 @@ private void doChunkUploadTest(int contentLength, boolean chunkUpload, String up postBlobHandler.handle(request, restResponseChannel, future::done); if (errorChecker == null) { future.get(TIMEOUT_SECS, TimeUnit.SECONDS); + assertNull(request.getArgs().get(BLOB_ID_IS_SERVER_GENERATED)); assertEquals("Unexpected converted ID", CONVERTED_ID, restResponseChannel.getHeader(RestUtils.Headers.LOCATION)); Object metadata = request.getArgs().get(RestUtils.InternalKeys.SIGNED_ID_METADATA_KEY); if (chunkUpload) { @@ -694,6 +695,7 @@ private void stitchBlobAndVerify(byte[] requestBody, List expectedSti postBlobHandler.handle(request, restResponseChannel, future::done); if (errorChecker == null) { future.get(TIMEOUT_SECS, TimeUnit.SECONDS); + assertNull(request.getArgs().get(BLOB_ID_IS_SERVER_GENERATED)); assertEquals("Unexpected converted ID", CONVERTED_ID, restResponseChannel.getHeader(RestUtils.Headers.LOCATION)); InMemoryRouter.InMemoryBlob blob = router.getActiveBlobs().get(idConverterFactory.lastInput); assertEquals("List of chunks stitched does not match expected", expectedStitchedChunks, blob.getStitchedChunks()); diff --git a/ambry-frontend/src/test/java/com/github/ambry/frontend/S3MultipartUploadTest.java b/ambry-frontend/src/test/java/com/github/ambry/frontend/S3MultipartUploadTest.java index 8f5a532656..6d5c683b29 100644 --- a/ambry-frontend/src/test/java/com/github/ambry/frontend/S3MultipartUploadTest.java +++ b/ambry-frontend/src/test/java/com/github/ambry/frontend/S3MultipartUploadTest.java @@ -143,6 +143,7 @@ public void multiPartUploadTest() throws Exception { FutureResult putResult = new FutureResult<>(); s3PutHandler.handle(request, restResponseChannel, putResult::done); putResult.get(); + assertNull(request.getArgs().get(InternalKeys.BLOB_ID_IS_SERVER_GENERATED)); String etag1 = (String) restResponseChannel.getHeader(Headers.ETAG); String location1 = (String) restResponseChannel.getHeader(Headers.LOCATION); assertEquals("Mismatch on response status", ResponseStatus.Ok, restResponseChannel.getStatus()); @@ -196,6 +197,7 @@ public void multiPartUploadTest() throws Exception { // Verify Upsert header is set by default for S3 multipart uploads. assertTrue("Upsert header must be present", request.getArgs().containsKey(NAMED_UPSERT)); readableStreamChannel = postResult.get(); + assertNull(request.getArgs().get(InternalKeys.BLOB_ID_IS_SERVER_GENERATED)); byteBuffer = ((ByteBufferReadableStreamChannel) readableStreamChannel).getContent(); byte[] byteArray = byteBuffer.array(); CompleteMultipartUploadResult completeMultipartUploadResult = diff --git a/ambry-frontend/src/test/java/com/github/ambry/frontend/S3PutHandlerTest.java b/ambry-frontend/src/test/java/com/github/ambry/frontend/S3PutHandlerTest.java index 48539cb5b2..84a87fcf8a 100644 --- a/ambry-frontend/src/test/java/com/github/ambry/frontend/S3PutHandlerTest.java +++ b/ambry-frontend/src/test/java/com/github/ambry/frontend/S3PutHandlerTest.java @@ -104,6 +104,7 @@ public void putBlobsTestHelper(boolean provideContentType) throws Exception { restResponseChannel.setStatus(ResponseStatus.Created); s3PutHandler.handle(request, restResponseChannel, putResult::done); putResult.get(); + assertNull(request.getArgs().get(BLOB_ID_IS_SERVER_GENERATED)); // Verify Upsert header is set by default for S3 uploads. assertTrue("Upsert header must be present", request.getArgs().containsKey(NAMED_UPSERT)); diff --git a/ambry-router/src/main/java/com/github/ambry/router/NonBlockingRouter.java b/ambry-router/src/main/java/com/github/ambry/router/NonBlockingRouter.java index ddd6d98eb5..b3ae7ac395 100644 --- a/ambry-router/src/main/java/com/github/ambry/router/NonBlockingRouter.java +++ b/ambry-router/src/main/java/com/github/ambry/router/NonBlockingRouter.java @@ -1088,6 +1088,7 @@ private Callback createIdConverterCallbackForPutAndStitch(RestRequest re blobProperties.setBlobSize(restRequest.getBlobBytesReceived()); } // Call idConverter.convert after putBlob succeeds + restRequest.setArg(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED, true); idConverter.convert(restRequest, blobId, blobProperties, callback); } }; diff --git a/ambry-router/src/test/java/com/github/ambry/router/UploadIdConversionTest.java b/ambry-router/src/test/java/com/github/ambry/router/UploadIdConversionTest.java new file mode 100644 index 0000000000..4df2017fd7 --- /dev/null +++ b/ambry-router/src/test/java/com/github/ambry/router/UploadIdConversionTest.java @@ -0,0 +1,132 @@ +/* + * Copyright 2026 LinkedIn Corp. All rights reserved. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + */ +package com.github.ambry.router; + +import com.github.ambry.commons.Callback; +import com.github.ambry.commons.LoggingNotificationSystem; +import com.github.ambry.frontend.IdConverter; +import com.github.ambry.frontend.IdConverterFactory; +import com.github.ambry.messageformat.MessageFormatRecord; +import com.github.ambry.rest.MockRestRequest; +import com.github.ambry.rest.RequestPath; +import com.github.ambry.rest.RestMethod; +import com.github.ambry.rest.RestRequest; +import com.github.ambry.rest.RestServiceErrorCode; +import com.github.ambry.rest.RestServiceException; +import com.github.ambry.rest.RestUtils; +import com.github.ambry.utils.Utils; +import java.util.Arrays; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CompletableFuture; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; +import org.json.JSONObject; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.Parameterized; + +import static com.github.ambry.router.RouterTestHelpers.*; +import static com.github.ambry.utils.TestUtils.*; +import static org.junit.Assert.*; +import static org.mockito.Mockito.*; + + +@RunWith(Parameterized.class) +public class UploadIdConversionTest extends NonBlockingRouterTestBase { + public UploadIdConversionTest(boolean encrypted) throws Exception { + super(encrypted, MessageFormatRecord.Metadata_Content_Version_V3, false); + } + + @Parameterized.Parameters + public static List data() { + return Arrays.asList(new Object[][]{{false}, {true}}); + } + + @Test + public void testGeneratedBlobIdConversionErrors() throws Exception { + IdConverterFactory factory = mock(IdConverterFactory.class); + IdConverter converter = mock(IdConverter.class); + when(factory.getIdConverter()).thenReturn(converter); + setRouterWithIdConverterFactory(getNonBlockingRouterProperties(localDcName), + new MockServerLayout(mockClusterMap), new LoggingNotificationSystem(), factory); + setOperationParams(); + String chunkId = router.putBlob(putBlobProperties, putUserMetadata, putChannel, putOptionsForChunkedUpload) + .get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS); + for (boolean stitch : new boolean[]{false, true}) { + for (Exception failure : new Exception[]{ + null, + new RouterException("invalid ID", RouterErrorCode.InvalidBlobId), + new RestServiceException("invalid input", RestServiceErrorCode.BadRequest)}) { + CompletableFuture> pendingConversion = new CompletableFuture<>(); + FutureResult conversion = new FutureResult<>(); + when(converter.convert(any(), anyString(), any(), any())).thenAnswer(invocation -> { + pendingConversion.complete(invocation.getArgument(3)); + return conversion; + }); + setOperationParams(); + try (RestRequest request = createUploadRequest(stitch)) { + FutureResult result = new FutureResult<>(); + AtomicInteger callbackCount = new AtomicInteger(); + Callback callback = (id, error) -> { + callbackCount.incrementAndGet(); + result.done(id, error); + }; + Future stored = stitch ? router.stitchBlob(request, putBlobProperties, putUserMetadata, + Collections.singletonList(new ChunkInfo(chunkId, PUT_CONTENT_SIZE, Utils.Infinite_Time, null)), + null, callback, null) + : router.putBlob(request, putBlobProperties, putUserMetadata, putChannel, + new PutBlobOptionsBuilder().build(), callback, null); + String blobId = stored.get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS); + assertFalse("Upload callback must wait for ID conversion", result.isDone()); + conversion.done(failure == null ? blobId : null, failure); + Callback conversionCallback = pendingConversion.get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS); + assertEquals(Boolean.TRUE, request.getArgs().get(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED)); + conversionCallback.onCompletion(failure == null ? blobId : null, failure); + if (failure == null) { + assertEquals(blobId, result.get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS)); + } else { + assertException(ExecutionException.class, () -> result.get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS), + e -> assertSame(failure, e.getCause())); + } + assertEquals(1, callbackCount.get()); + GetBlobResult blob = router.getBlob(blobId, new GetBlobOptionsBuilder().build()) + .get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS); + blob.getBlobDataChannel().close(); + } + } + } + clearInvocations(converter); + FutureResult result = new FutureResult<>(); + try (RestRequest request = createUploadRequest(true)) { + router.stitchBlob(request, putBlobProperties, putUserMetadata, + Collections.singletonList(new ChunkInfo("invalid-client-id", 1, Utils.Infinite_Time, null)), + null, result::done, null); + assertException(ExecutionException.class, () -> result.get(AWAIT_TIMEOUT_MS, TimeUnit.MILLISECONDS), + e -> assertEquals(RouterErrorCode.InvalidBlobId, ((RouterException) e.getCause()).getErrorCode())); + verifyNoInteractions(converter); + assertNull(request.getArgs().get(RestUtils.InternalKeys.BLOB_ID_IS_SERVER_GENERATED)); + } + } + + private RestRequest createUploadRequest(boolean stitch) throws Exception { + JSONObject headers = new JSONObject().put(RestUtils.Headers.UPLOAD_NAMED_BLOB_MODE, stitch ? RestUtils.STITCH : ""); + RestRequest request = new MockRestRequest(new JSONObject().put(MockRestRequest.REST_METHOD_KEY, RestMethod.PUT.name()) + .put(MockRestRequest.URI_KEY, "/named/account/container/blob").put(MockRestRequest.HEADERS_KEY, headers), null); + request.setArg(RestUtils.InternalKeys.REQUEST_PATH, RequestPath.parse(request, Collections.emptyList(), null)); + return request; + } +}