@Test
  public void testReserveMetadataBlobId() throws Exception {
    RouterConfig[] routerConfigs = new RouterConfig[] { createRouterConfig(true), routerConfig};
    int metadataIdNotReservedBaseCount = 0;
    int stitchedBlobMetadataIdDeserErrorBaseCount = 0;
    int stitchedBlobMetadataIdMismatchBaseCount = 0;
    int reservedMetadataPassedInForNonChunkedUploadBaseCount = 0;
    int noReservedMetadataForChunkedUploadBaseCount = 0;

    for (RouterConfig routerConfig : routerConfigs) {
      BlobProperties blobProperties =
          new BlobProperties(-1, "serviceId", "memberId", "contentType", false, Utils.Infinite_Time, Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), false, null, null, null);
      byte[] userMetadata = new byte[10];
      FutureResult<String> future = new FutureResult<>();
      MockNetworkClient mockNetworkClient = new MockNetworkClient();
      BlobId reservedMetadataId =
          new BlobId(BlobId.BLOB_ID_V6, BlobId.BlobIdType.NATIVE, (byte) 0, (short) 1, (short) 1, new MockPartitionId(),
              false, BlobId.BlobDataType.METADATA);
      BlobProperties blobPropertiesWithId =
          new BlobProperties(-1, "serviceId", "memberId", "contentType", false, Utils.Infinite_Time, Utils.getRandomShort(TestUtils.RANDOM), Utils.getRandomShort(TestUtils.RANDOM), false, null, null, null,
              reservedMetadataId.getID());
      BlobId altReservedMetadataId =
          new BlobId(BlobId.BLOB_ID_V6, BlobId.BlobIdType.NATIVE, (byte) 0, (short) 1, (short) 1, new MockPartitionId(),
              true, BlobId.BlobDataType.METADATA);
      ReservedMetadataIdMetrics reservedMetadataIdMetrics = ReservedMetadataIdMetrics.getReservedMetadataIdMetrics(routerMetrics.getMetricRegistry());

      // if stitched blob and chunks don't have reserved metadata chunk, then metadata chunk id should not be reserved.
      List<ChunkInfo> chunksToStitch =
          RouterTestHelpers.buildChunkList(mockClusterMap, BlobId.BlobDataType.DATACHUNK, Utils.Infinite_Time,
              LongStream.of(10, 10, 10));
      PutOperation op =
          PutOperation.forStitching(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(), new InMemAccountService(true, false), userMetadata, chunksToStitch, future, null,
              new RouterCallback(mockNetworkClient, new ArrayList<>()), null, null, null, time, blobProperties,
              MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertNull(op.getReservedMetadataId());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 1, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 1,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());

      // if stitched blob and chunks have reserved metadata chunk, then the metadata chunk id should be reserved.
      chunksToStitch =
          RouterTestHelpers.buildChunkList(mockClusterMap, BlobId.BlobDataType.DATACHUNK, Utils.Infinite_Time,
              LongStream.of(10, 10, 10), reservedMetadataId.getID());
      op = PutOperation.forStitching(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(), new InMemAccountService(true, false), userMetadata, chunksToStitch, future, null,
          new RouterCallback(mockNetworkClient, new ArrayList<>()), null, null, null, time, blobProperties,
          MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertEquals(reservedMetadataId, op.getReservedMetadataId());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 1, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 1,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());
      if (routerConfig.routerReservedMetadataEnabled) {
        // If the reserved metadata is enabled check that the operation's metadata partition id is set to the reserved
        // metadata partition id.
        op.startOperation();
        op.poll(requestRegistrationCallback);
        Assert.assertEquals(((PutOperation.PutChunk) op.getMetadataPutChunk()).getPartitionId(), reservedMetadataId.getPartition());
        Assert.assertEquals(((PutOperation.PutChunk) op.getMetadataPutChunk()).chunkBlobId, reservedMetadataId);
      }

      // if stitched blob and chunks have reserved metadata chunk, but the reserved chunk has mismatch, then the metadata
      // chunk id should not be reserved.
      chunksToStitch =
          RouterTestHelpers.buildChunkList(mockClusterMap, BlobId.BlobDataType.DATACHUNK, Utils.Infinite_Time,
              LongStream.of(10, 10, 10), reservedMetadataId.getID());
      ChunkInfo chunkInfo = new ChunkInfo(chunksToStitch.get(1).getBlobId(), chunksToStitch.get(1).getChunkSizeInBytes(),
          chunksToStitch.get(1).getExpirationTimeInMs(), altReservedMetadataId.getID());
      chunksToStitch.set(1, chunkInfo);
      op = PutOperation.forStitching(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(), new InMemAccountService(true, false), userMetadata, chunksToStitch, future, null,
          new RouterCallback(mockNetworkClient, new ArrayList<>()), null, null, null, time, blobProperties,
          MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertNull(op.getReservedMetadataId());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 2, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 1,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount + 1,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());

      chunksToStitch =
          RouterTestHelpers.buildChunkList(mockClusterMap, BlobId.BlobDataType.DATACHUNK, Utils.Infinite_Time,
              LongStream.of(10, 10, 10), reservedMetadataId.getID());
      chunkInfo = new ChunkInfo(chunksToStitch.get(1).getBlobId(), chunksToStitch.get(1).getChunkSizeInBytes(),
          chunksToStitch.get(1).getExpirationTimeInMs(), null);
      chunksToStitch.set(1, chunkInfo);
      op = PutOperation.forStitching(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(), new InMemAccountService(true, false), userMetadata, chunksToStitch, future, null,
          new RouterCallback(mockNetworkClient, new ArrayList<>()), null, null, null, time, blobProperties,
          MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertNull(op.getReservedMetadataId());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 3, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 1,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());

      chunksToStitch =
          RouterTestHelpers.buildChunkList(mockClusterMap, BlobId.BlobDataType.DATACHUNK, Utils.Infinite_Time,
              LongStream.of(10, 10, 10), reservedMetadataId.getID());
      chunkInfo = new ChunkInfo(chunksToStitch.get(0).getBlobId(), chunksToStitch.get(0).getChunkSizeInBytes(),
          chunksToStitch.get(0).getExpirationTimeInMs(), null);
      chunksToStitch.set(0, chunkInfo);
      op = PutOperation.forStitching(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(), new InMemAccountService(true, false), userMetadata, chunksToStitch, future, null,
          new RouterCallback(mockNetworkClient, new ArrayList<>()), null, null, null, time, blobProperties,
          MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertNull(op.getReservedMetadataId());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 4, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());

      // if regular non-chunked blob and blob properties doesn't have reserved metadata id, then a metadata id should be reserved.
      int numChunks = routerConfig.routerMaxInMemPutChunks + 1;
      byte[] content = new byte[chunkSize * numChunks];
      ReadableStreamChannel channel = new ByteBufferReadableStreamChannel(ByteBuffer.wrap(content));
      op = PutOperation.forUpload(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(),
          new InMemAccountService(true, false), userMetadata, channel, PutBlobOptions.DEFAULT, new FutureResult<>(),
          null, new RouterCallback(new MockNetworkClient(), new ArrayList<>()), null, null, null, null, time,
          blobProperties, MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertNotNull(op.getReservedMetadataId());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 4, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());

      // if regular non-chunked blob and blob properties have reserved metadata id, then a metadata id should not be reserved.
      channel = new ByteBufferReadableStreamChannel(ByteBuffer.wrap(content));
      op = PutOperation.forUpload(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(),
          new InMemAccountService(true, false), userMetadata, channel, PutBlobOptions.DEFAULT, new FutureResult<>(),
          null, new RouterCallback(new MockNetworkClient(), new ArrayList<>()), null, null, null, null, time,
          blobPropertiesWithId, MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertNull(op.getReservedMetadataId());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 5, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount + 1,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());

      // if stitched upload and blob properties has invalid reserved metadata chunk then metadata id should be null.
      List<ChunkInfo> invalidChunksToStitch =
          RouterTestHelpers.buildChunkList(mockClusterMap, BlobId.BlobDataType.DATACHUNK, Utils.Infinite_Time,
              LongStream.of(10, 10, 10), "test");
      op = PutOperation.forStitching(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(), new InMemAccountService(true, false), userMetadata, invalidChunksToStitch, future, null,
          new RouterCallback(mockNetworkClient, new ArrayList<>()), null, null, null, time, blobProperties,
          MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertNull(op.getReservedMetadataId());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 6, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 3,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount + 1,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());

      // if chunked upload and blob properties doesn't have reserved metadata id, then a metadata id should not be reserved.
      content = new byte[chunkSize];
      channel = new ByteBufferReadableStreamChannel(ByteBuffer.wrap(content));
      op = PutOperation.forUpload(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(),
          new InMemAccountService(true, false), userMetadata, channel, new PutBlobOptionsBuilder().chunkUpload(true).build(), new FutureResult<>(),
          null, new RouterCallback(new MockNetworkClient(), new ArrayList<>()), null, null, null, null, time,
          blobProperties, MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertNull(op.getReservedMetadataId());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 7, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 3,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount + 1,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());
      Assert.assertEquals(noReservedMetadataForChunkedUploadBaseCount + 1,
          reservedMetadataIdMetrics.noReservedMetadataForChunkedUploadCount.getCount());

      // if chunked upload and blob properties have reserved metadata id, then a metadata id should be reserved.
      channel = new ByteBufferReadableStreamChannel(ByteBuffer.wrap(content));
      op = PutOperation.forUpload(routerConfig, routerMetrics, mockClusterMap, new LoggingNotificationSystem(),
          new InMemAccountService(true, false), userMetadata, channel, new PutBlobOptionsBuilder().chunkUpload(true).build(), new FutureResult<>(),
          null, new RouterCallback(new MockNetworkClient(), new ArrayList<>()), null, null, null, null, time,
          blobPropertiesWithId, MockClusterMap.DEFAULT_PARTITION_CLASS, quotaChargeCallback, compressionService);
      Assert.assertEquals(blobPropertiesWithId.getReservedMetadataBlobId(), op.getReservedMetadataId().getID());
      Assert.assertEquals(
          metadataIdNotReservedBaseCount + 7, reservedMetadataIdMetrics.metadataIdNotReservedCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdDeserErrorBaseCount + 3,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdDeserErrorCount.getCount());
      Assert.assertEquals(stitchedBlobMetadataIdMismatchBaseCount + 2,
          reservedMetadataIdMetrics.stitchedBlobMetadataIdMismatchCount.getCount());
      Assert.assertEquals(reservedMetadataPassedInForNonChunkedUploadBaseCount + 1,
          reservedMetadataIdMetrics.reservedMetadataPassedInForNonChunkedUploadCount.getCount());
      Assert.assertEquals(noReservedMetadataForChunkedUploadBaseCount + 1,
          reservedMetadataIdMetrics.noReservedMetadataForChunkedUploadCount.getCount());

      metadataIdNotReservedBaseCount += 7;
      stitchedBlobMetadataIdMismatchBaseCount += 2;
      stitchedBlobMetadataIdDeserErrorBaseCount += 3;
      reservedMetadataPassedInForNonChunkedUploadBaseCount += 1;
      noReservedMetadataForChunkedUploadBaseCount += 1;
    }
  }
Ensure that errors while stitching blobs do not result in data chunk deletions.
@throws Exception
ambry-router/src/test/java/com/github/ambry/router/PutOperationTest.java